diff --git a/src/autoskillit/config/settings.py b/src/autoskillit/config/settings.py index 6ad80b02..2041d68a 100644 --- a/src/autoskillit/config/settings.py +++ b/src/autoskillit/config/settings.py @@ -11,7 +11,9 @@ from __future__ import annotations import dataclasses +import inspect import logging +import os import tempfile from dataclasses import dataclass, field from pathlib import Path @@ -193,6 +195,25 @@ class LinuxTracingConfig: log_dir: str = "" # empty = platform default (~/.local/share/autoskillit/logs on Linux) tmpfs_path: str = "/dev/shm" # RAM-backed tmpfs for crash-resilient streaming + def __post_init__(self) -> None: + if self.tmpfs_path != "/dev/shm" or not os.environ.get("PYTEST_CURRENT_TEST"): + return + # Only raise when called directly from test code — not from library machinery + # (e.g. AutomationConfig default_factory, from_dynaconf). We inspect the call + # frame two levels up: __post_init__ → __init__ (generated) → actual caller. + frame = inspect.currentframe() + init_frame = frame.f_back if frame is not None else None + caller = init_frame.f_back if init_frame is not None else None + if caller is not None and "/tests/" in (caller.f_code.co_filename or ""): + raise RuntimeError( + "LinuxTracingConfig.tmpfs_path is '/dev/shm' but PYTEST_CURRENT_TEST " + "is set — this test would write to the real shared tmpfs and pollute " + "production state. Override tmpfs_path with a test-local path, e.g.: " + "LinuxTracingConfig(tmpfs_path=str(tmp_path)). " + "Use the isolated_tracing_config fixture for new tests." + ) + del frame, init_frame, caller + @dataclass class McpResponseConfig: diff --git a/src/autoskillit/execution/linux_tracing.py b/src/autoskillit/execution/linux_tracing.py index b7a4612d..86aa95b4 100644 --- a/src/autoskillit/execution/linux_tracing.py +++ b/src/autoskillit/execution/linux_tracing.py @@ -18,7 +18,9 @@ import dataclasses import json +import os import sys +import tempfile from collections.abc import AsyncIterator from dataclasses import dataclass, field from datetime import UTC, datetime, timedelta @@ -29,9 +31,13 @@ import anyio.abc import psutil +from autoskillit.core import get_logger + if TYPE_CHECKING: from autoskillit.config import LinuxTracingConfig +logger = get_logger(__name__) + LINUX_TRACING_AVAILABLE = sys.platform == "linux" @@ -58,6 +64,59 @@ def read_starttime_ticks(pid: int) -> int | None: return None +@dataclass(frozen=True) +class TraceEnrollmentRecord: + """Identity triple written atomically at trace-open time. + + (boot_id, pid, starttime_ticks) together form a collision-resistant identity: + - boot_id rejects pre-reboot stale files + - starttime_ticks detects PID recycling + """ + + schema_version: int # always 1 + pid: int + boot_id: str | None # read_boot_id(); None if /proc unavailable + starttime_ticks: int | None # read_starttime_ticks(pid); None if unavailable + session_id: str # caller-provided; "" if not yet resolved + enrolled_at: str # ISO 8601 UTC + kitchen_id: str # "" + order_id: str # "" + + +def _write_enrollment_atomic(path: Path, record: TraceEnrollmentRecord) -> None: + """Write enrollment sidecar atomically using tempfile + os.replace.""" + content = json.dumps(dataclasses.asdict(record)) + fd, tmp = tempfile.mkstemp(dir=path.parent, suffix=".tmp") + try: + with os.fdopen(fd, "w", encoding="utf-8") as f: + f.write(content) + os.replace(tmp, path) + except Exception: + try: + os.unlink(tmp) + except OSError: + pass + raise + + +def read_enrollment(path: Path) -> TraceEnrollmentRecord | None: + """Read and validate an enrollment sidecar. Returns None on any error.""" + try: + data = json.loads(path.read_text(encoding="utf-8")) + return TraceEnrollmentRecord( + schema_version=data["schema_version"], + pid=data["pid"], + boot_id=data.get("boot_id"), + starttime_ticks=data.get("starttime_ticks"), + session_id=data.get("session_id", ""), + enrolled_at=data.get("enrolled_at", ""), + kitchen_id=data.get("kitchen_id", ""), + order_id=data.get("order_id", ""), + ) + except (OSError, KeyError, TypeError, ValueError, json.JSONDecodeError): + return None + + @dataclass(frozen=True) class ProcSnapshot: """Point-in-time snapshot of process state.""" @@ -200,6 +259,7 @@ class LinuxTracingHandle: _snapshots: list[ProcSnapshot] = field(default_factory=list) _trace_path: Path | None = field(default=None) _trace_file: IO[str] | None = field(default=None) + _enrollment_path: Path | None = field(default=None) def stop(self) -> list[ProcSnapshot]: """Stop tracing, flush and close the trace file, return accumulated snapshots.""" @@ -212,6 +272,14 @@ def stop(self) -> list[ProcSnapshot]: except OSError: pass self._trace_file = None + if self._trace_path is not None: + # Intentional: stop() cleans up its own trace file. Crash-recovery only reads + # files left behind by processes that never called stop() — so this is correct. + self._trace_path.unlink(missing_ok=True) + self._trace_path = None + if self._enrollment_path is not None: + self._enrollment_path.unlink(missing_ok=True) + self._enrollment_path = None return list(self._snapshots) @@ -219,6 +287,10 @@ def start_linux_tracing( pid: int, config: LinuxTracingConfig, tg: anyio.abc.TaskGroup | None, + *, + session_id: str = "", + kitchen_id: str = "", + order_id: str = "", ) -> LinuxTracingHandle | None: """Start Linux tracing if all gates pass. Returns handle or None.""" if not LINUX_TRACING_AVAILABLE or not config.enabled: @@ -240,6 +312,25 @@ def start_linux_tracing( handle._trace_path = None handle._trace_file = None + # Write enrollment sidecar atomically for crash-recovery identity contract + enrollment_path = tmpfs / f"autoskillit_enrollment_{pid}.json" + try: + record = TraceEnrollmentRecord( + schema_version=1, + pid=pid, + boot_id=read_boot_id(), + starttime_ticks=read_starttime_ticks(pid), + session_id=session_id, + enrolled_at=datetime.now(UTC).isoformat(), + kitchen_id=kitchen_id, + order_id=order_id, + ) + _write_enrollment_atomic(enrollment_path, record) + handle._enrollment_path = enrollment_path + except OSError as e: + logger.warning("Failed to write enrollment sidecar for pid %d: %s", pid, e) + handle._enrollment_path = None + async def _run_monitor() -> None: with scope: async for snap in proc_monitor(pid, config.proc_interval): diff --git a/src/autoskillit/execution/session_log.py b/src/autoskillit/execution/session_log.py index 891e0f1f..95685de3 100644 --- a/src/autoskillit/execution/session_log.py +++ b/src/autoskillit/execution/session_log.py @@ -17,8 +17,15 @@ from pathlib import Path from typing import Any +import psutil + from autoskillit.core import atomic_write, claude_code_log_path, get_logger from autoskillit.execution.anomaly_detection import detect_anomalies +from autoskillit.execution.linux_tracing import ( + read_boot_id, + read_enrollment, + read_starttime_ticks, +) logger = get_logger(__name__) @@ -315,6 +322,7 @@ def recover_crashed_sessions(tmpfs_path: str = "/dev/shm", log_dir: str = "") -> return 0 count = 0 + current_boot_id = read_boot_id() for trace_file in sorted(tmpfs.glob("autoskillit_trace_*.jsonl")): # Skip files modified within the last 30 seconds — may be active try: @@ -324,7 +332,35 @@ def recover_crashed_sessions(tmpfs_path: str = "/dev/shm", log_dir: str = "") -> if age_seconds < 30: continue - # Read snapshots + # Extract PID from filename: autoskillit_trace_{pid}.jsonl + try: + pid = int(trace_file.stem.split("_")[-1]) + except (ValueError, IndexError): + pid = -1 + + # Gate 1: Enrollment sidecar must exist — no sidecar means alien/test file + enrollment_path = tmpfs / f"autoskillit_enrollment_{pid}.json" + enrollment = read_enrollment(enrollment_path) + if enrollment is None: + logger.debug("Skipping %s: no enrollment sidecar", trace_file.name) + continue + + # Gate 2: Boot ID must match current boot — mismatch means pre-reboot stale file + if current_boot_id and enrollment.boot_id and enrollment.boot_id != current_boot_id: + logger.debug("Skipping %s: boot_id mismatch", trace_file.name) + trace_file.unlink(missing_ok=True) + enrollment_path.unlink(missing_ok=True) + continue + + # Gate 3: PID liveness + starttime_ticks identity + if psutil.pid_exists(pid): + current_ticks = read_starttime_ticks(pid) + if current_ticks is not None and current_ticks == enrollment.starttime_ticks: + logger.debug("Skipping %s: PID %d still alive", trace_file.name, pid) + continue + # PID recycled — original process is gone, treat as crash + + # All gates passed — read snapshots and emit crashed row snapshots: list[dict[str, object]] = [] try: for line in trace_file.read_text().splitlines(): @@ -335,12 +371,6 @@ def recover_crashed_sessions(tmpfs_path: str = "/dev/shm", log_dir: str = "") -> except OSError: continue - # Extract PID from filename: autoskillit_trace_{pid}.jsonl - try: - pid = int(trace_file.stem.split("_")[-1]) - except (ValueError, IndexError): - pid = 0 - # Compute start_ts from file mtime try: mtime_ts = datetime.fromtimestamp(trace_file.stat().st_mtime, tz=UTC).isoformat() @@ -365,10 +395,8 @@ def recover_crashed_sessions(tmpfs_path: str = "/dev/shm", log_dir: str = "") -> logger.debug("recover_crashed_sessions: failed to finalize %s", trace_file) continue - try: - trace_file.unlink() - except OSError: - pass + trace_file.unlink(missing_ok=True) + enrollment_path.unlink(missing_ok=True) count += 1 diff --git a/tests/conftest.py b/tests/conftest.py index add85e59..772a0e25 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -255,6 +255,7 @@ def tool_ctx(monkeypatch, tmp_path): ctx = make_context(AutomationConfig(), runner=mock_runner, plugin_dir=str(tmp_path)) ctx.gate = DefaultGateState(enabled=True) ctx.config.linux_tracing.log_dir = str(tmp_path / "session_logs") + ctx.config.linux_tracing.tmpfs_path = str(tmp_path / "shm") # Anchor temp_dir to tmp_path so server tools that read from ctx.temp_dir # (e.g. _apply_triage_gate's staleness cache) write under the per-test # tmp directory rather than the cwd captured at fixture-init time. diff --git a/tests/execution/conftest.py b/tests/execution/conftest.py index ed56c65e..50d31a67 100644 --- a/tests/execution/conftest.py +++ b/tests/execution/conftest.py @@ -2,8 +2,13 @@ from __future__ import annotations +import pathlib import textwrap +import pytest + +from autoskillit.config.settings import LinuxTracingConfig + # Simulates Claude CLI process that writes a result line then hangs. # Used by test_process_channel_b.py and test_process_monitor.py. WRITE_RESULT_THEN_HANG_SCRIPT = textwrap.dedent("""\ @@ -14,3 +19,13 @@ sys.stdout.flush() time.sleep(3600) """) + + +@pytest.fixture +def isolated_tracing_config(tmp_path: pathlib.Path) -> LinuxTracingConfig: + """Pre-isolated LinuxTracingConfig for tracing tests. + Always writes to a tmp_path subdir, never to the real /dev/shm. + Use this fixture for all new tests that need a LinuxTracingConfig.""" + shm = tmp_path / "shm" + shm.mkdir(parents=True, exist_ok=True) + return LinuxTracingConfig(enabled=True, proc_interval=0.05, tmpfs_path=str(shm)) diff --git a/tests/execution/test_linux_tracing.py b/tests/execution/test_linux_tracing.py index af01a3de..46f6dc6b 100644 --- a/tests/execution/test_linux_tracing.py +++ b/tests/execution/test_linux_tracing.py @@ -78,7 +78,7 @@ def test_read_proc_snapshot_has_all_fields(): @pytest.mark.anyio -async def test_tracing_handle_accumulates_snapshots(): +async def test_tracing_handle_accumulates_snapshots(tmp_path): """LinuxTracingHandle accumulates snapshots during monitoring.""" import subprocess @@ -88,7 +88,7 @@ async def test_tracing_handle_accumulates_snapshots(): from autoskillit.execution.linux_tracing import start_linux_tracing proc = subprocess.Popen(["sleep", "2"]) - cfg = LinuxTracingConfig(enabled=True, proc_interval=0.1) + cfg = LinuxTracingConfig(enabled=True, proc_interval=0.1, tmpfs_path=str(tmp_path)) async with anyio.create_task_group() as tg: handle = start_linux_tracing(pid=proc.pid, config=cfg, tg=tg) @@ -104,7 +104,7 @@ async def test_tracing_handle_accumulates_snapshots(): @pytest.mark.anyio -async def test_tracing_handle_stop_returns_snapshots(): +async def test_tracing_handle_stop_returns_snapshots(tmp_path): """stop() returns the accumulated snapshot list.""" import os @@ -113,7 +113,7 @@ async def test_tracing_handle_stop_returns_snapshots(): from autoskillit.config import LinuxTracingConfig from autoskillit.execution.linux_tracing import start_linux_tracing - cfg = LinuxTracingConfig(enabled=True, proc_interval=0.1) + cfg = LinuxTracingConfig(enabled=True, proc_interval=0.1, tmpfs_path=str(tmp_path)) async with anyio.create_task_group() as tg: handle = start_linux_tracing(pid=os.getpid(), config=cfg, tg=tg) @@ -140,13 +140,13 @@ def test_linux_tracing_unavailable_on_non_linux(): assert LINUX_TRACING_AVAILABLE is False -def test_noop_on_non_linux(monkeypatch): +def test_noop_on_non_linux(monkeypatch, tmp_path): """start_linux_tracing is a no-op when LINUX_TRACING_AVAILABLE is False.""" from autoskillit.config import LinuxTracingConfig from autoskillit.execution import linux_tracing monkeypatch.setattr(linux_tracing, "LINUX_TRACING_AVAILABLE", False) - cfg = LinuxTracingConfig(enabled=True, proc_interval=1.0) + cfg = LinuxTracingConfig(enabled=True, proc_interval=1.0, tmpfs_path=str(tmp_path)) result = linux_tracing.start_linux_tracing(pid=1, config=cfg, tg=None) assert result is None @@ -219,9 +219,14 @@ async def test_streaming_writes_each_snapshot_as_jsonl(tmp_path): await anyio.sleep(0.2) tg.cancel_scope.cancel() + # Save trace path and flush before stop() deletes the file on clean exit + trace_path = handle._trace_path + assert trace_path is not None + if handle._trace_file is not None: + handle._trace_file.flush() + content = trace_path.read_text() snapshots = handle.stop() - assert handle._trace_path is not None - lines = handle._trace_path.read_text().strip().split("\n") + lines = content.strip().split("\n") assert len(lines) >= 1 for line in lines: record = json.loads(line) @@ -299,7 +304,7 @@ def test_proc_snapshot_has_captured_at_field(): @pytest.mark.anyio -async def test_proc_monitor_snapshots_have_distinct_timestamps(): +async def test_proc_monitor_snapshots_have_distinct_timestamps(tmp_path): """Consecutive snapshots from proc_monitor must have distinct captured_at values.""" import os @@ -308,7 +313,7 @@ async def test_proc_monitor_snapshots_have_distinct_timestamps(): from autoskillit.config import LinuxTracingConfig from autoskillit.execution.linux_tracing import start_linux_tracing - config = LinuxTracingConfig(proc_interval=0.05) + config = LinuxTracingConfig(proc_interval=0.05, tmpfs_path=str(tmp_path)) async with anyio.create_task_group() as tg: handle = start_linux_tracing(os.getpid(), config, tg) await anyio.sleep(0.2) @@ -348,3 +353,98 @@ async def test_proc_monitor_persists_psutil_process_for_cpu_percent(): finally: proc.kill() proc.wait() + + +@pytest.mark.anyio +async def test_start_linux_tracing_writes_enrollment_sidecar(tmp_path): + """start_linux_tracing must write autoskillit_enrollment_{pid}.json immediately.""" + import anyio + + from autoskillit.config import LinuxTracingConfig + from autoskillit.execution.linux_tracing import start_linux_tracing + + cfg = LinuxTracingConfig(enabled=True, proc_interval=0.1, tmpfs_path=str(tmp_path)) + async with anyio.create_task_group() as tg: + proc = await anyio.open_process(["sleep", "2"]) + handle = start_linux_tracing(proc.pid, cfg, tg) + assert handle is not None + enrollment = tmp_path / f"autoskillit_enrollment_{proc.pid}.json" + assert enrollment.exists() + data = json.loads(enrollment.read_text()) + assert data["schema_version"] == 1 + assert data["pid"] == proc.pid + try: + handle.stop() + finally: + proc.kill() + assert not enrollment.exists(), "Enrollment must be deleted by stop()" + + +# --- LinuxTracingConfig guard tests --- + + +def test_tracing_config_rejects_dev_shm_in_test_env(monkeypatch): + """LinuxTracingConfig.__post_init__ must raise when tmpfs_path is /dev/shm + and PYTEST_CURRENT_TEST env var is set.""" + from autoskillit.config import LinuxTracingConfig + + monkeypatch.setenv( + "PYTEST_CURRENT_TEST", + "tests/execution/test_linux_tracing.py::fake_test", + ) + with pytest.raises(RuntimeError, match="tmpfs_path|PYTEST_CURRENT_TEST"): + LinuxTracingConfig(tmpfs_path="/dev/shm") + + +def test_tracing_config_allows_custom_tmpfs_in_test_env(monkeypatch, tmp_path): + """LinuxTracingConfig must not raise when a non-/dev/shm path is provided.""" + from autoskillit.config import LinuxTracingConfig + + monkeypatch.setenv( + "PYTEST_CURRENT_TEST", + "tests/execution/test_linux_tracing.py::fake_test", + ) + cfg = LinuxTracingConfig(tmpfs_path=str(tmp_path)) # must not raise + assert cfg.tmpfs_path == str(tmp_path) + + +def test_tracing_config_allows_dev_shm_outside_test_env(monkeypatch): + """LinuxTracingConfig must not raise for /dev/shm when not running under pytest.""" + from autoskillit.config import LinuxTracingConfig + + monkeypatch.delenv("PYTEST_CURRENT_TEST", raising=False) + cfg = LinuxTracingConfig(tmpfs_path="/dev/shm") # production path — must not raise + assert cfg.tmpfs_path == "/dev/shm" + + +def test_isolated_tracing_config_fixture_returns_non_dev_shm(isolated_tracing_config): + """The isolated_tracing_config fixture must return a config pointing to a + temp dir, not /dev/shm.""" + from pathlib import Path + + assert isolated_tracing_config.tmpfs_path != "/dev/shm" + assert Path(isolated_tracing_config.tmpfs_path).is_dir() + + +@pytest.mark.anyio +async def test_stop_unlinks_trace_and_enrollment(tmp_path): + """stop() must delete both trace JSONL and enrollment sidecar on clean exit.""" + import anyio + + from autoskillit.config import LinuxTracingConfig + from autoskillit.execution.linux_tracing import start_linux_tracing + + cfg = LinuxTracingConfig(enabled=True, proc_interval=0.1, tmpfs_path=str(tmp_path)) + async with anyio.create_task_group() as tg: + proc = await anyio.open_process(["sleep", "2"]) + handle = start_linux_tracing(proc.pid, cfg, tg) + assert handle is not None + trace = tmp_path / f"autoskillit_trace_{proc.pid}.jsonl" + enrollment = tmp_path / f"autoskillit_enrollment_{proc.pid}.json" + assert trace.exists() + try: + handle.stop() + finally: + proc.kill() + assert not trace.exists(), "Trace file must be deleted on clean stop()" + assert not enrollment.exists(), "Enrollment sidecar must be deleted on clean stop()" diff --git a/tests/execution/test_session_log.py b/tests/execution/test_session_log.py index e1b047a3..7ad74e8d 100644 --- a/tests/execution/test_session_log.py +++ b/tests/execution/test_session_log.py @@ -3,11 +3,15 @@ from __future__ import annotations import json +import os +import sys +import time from datetime import UTC, datetime from pathlib import Path import pytest +from autoskillit.execution.linux_tracing import read_boot_id, read_starttime_ticks from autoskillit.execution.session_log import ( flush_session_log, read_telemetry_clear_marker, @@ -387,15 +391,36 @@ def test_recover_crashed_sessions_skips_recent_files(tmp_path): def _write_old_trace(tmpfs: Path, filename: str, content: str) -> Path: - """Write a trace file and backdate its mtime to 60 seconds ago.""" - import time + """Write a trace file (backdated 60s) and its enrollment sidecar. + The enrollment sidecar uses the current boot_id so Gate 2 passes. + The PID embedded in the filename is expected to be dead (so Gate 3 passes). + """ trace = tmpfs / filename trace.write_text(content) old_mtime = time.time() - 60 - import os - os.utime(trace, (old_mtime, old_mtime)) + + # Write companion enrollment sidecar so Gate 1 passes + try: + pid = int(Path(filename).stem.split("_")[-1]) + except (ValueError, IndexError): + pid = 0 + enrollment = tmpfs / f"autoskillit_enrollment_{pid}.json" + enrollment.write_text( + json.dumps( + { + "schema_version": 1, + "pid": pid, + "boot_id": read_boot_id() or "", + "starttime_ticks": None, + "session_id": "", + "enrolled_at": "2026-01-01T00:00:00+00:00", + "kitchen_id": "", + "order_id": "", + } + ) + ) return trace @@ -811,3 +836,83 @@ def test_flush_session_log_order_id_defaults_to_empty(tmp_path): entry = json.loads((tmp_path / "sessions.jsonl").read_text().strip()) assert "order_id" in entry assert entry["order_id"] == "" + + +@pytest.mark.skipif(sys.platform != "linux", reason="Linux-only: uses /proc and boot_id") +def test_recover_crashed_sessions_skips_live_pid(tmp_path): + """A trace file whose enrolled PID is still alive must not be recovered.""" + tmpfs = tmp_path / "shm" + tmpfs.mkdir() + pid = os.getpid() + trace = tmpfs / f"autoskillit_trace_{pid}.jsonl" + enrollment = tmpfs / f"autoskillit_enrollment_{pid}.json" + trace.write_text("") + enrollment.write_text( + json.dumps( + { + "schema_version": 1, + "pid": pid, + "boot_id": read_boot_id() or "", + "starttime_ticks": read_starttime_ticks(pid), + "session_id": "", + "enrolled_at": datetime.now(UTC).isoformat(), + "kitchen_id": "", + "order_id": "", + } + ) + ) + os.utime(trace, (time.time() - 60,) * 2) + + count = recover_crashed_sessions(tmpfs_path=str(tmpfs), log_dir=str(tmp_path)) + + assert count == 0 + assert trace.exists(), "Trace file for alive PID must not be deleted" + assert enrollment.exists(), "Enrollment sidecar for alive PID must not be deleted" + + +def test_recover_crashed_sessions_skips_file_without_enrollment(tmp_path): + """A trace file with no enrollment sidecar must be skipped — it is not + an autoskillit-owned trace (e.g. a test artifact or alien file).""" + tmpfs = tmp_path / "shm" + tmpfs.mkdir() + trace = tmpfs / "autoskillit_trace_99997.jsonl" + trace.write_text("") + os.utime(trace, (time.time() - 60,) * 2) + + count = recover_crashed_sessions(tmpfs_path=str(tmpfs), log_dir=str(tmp_path)) + + assert count == 0 + assert trace.exists(), "Alien trace file must not be deleted" + + +def test_recover_crashed_sessions_skips_wrong_boot_id(tmp_path, monkeypatch): + """An enrollment sidecar with a different boot_id must be rejected.""" + monkeypatch.setattr( + "autoskillit.execution.session_log.read_boot_id", + lambda: "current-boot-id", + ) + tmpfs = tmp_path / "shm" + tmpfs.mkdir() + pid = 99996 + trace = tmpfs / f"autoskillit_trace_{pid}.jsonl" + enrollment = tmpfs / f"autoskillit_enrollment_{pid}.json" + trace.write_text("") + enrollment.write_text( + json.dumps( + { + "schema_version": 1, + "pid": pid, + "boot_id": "stale-boot-id", + "starttime_ticks": 1234, + "session_id": "", + "enrolled_at": "2026-01-01T00:00:00+00:00", + "kitchen_id": "", + "order_id": "", + } + ) + ) + os.utime(trace, (time.time() - 60,) * 2) + + count = recover_crashed_sessions(tmpfs_path=str(tmpfs), log_dir=str(tmp_path)) + + assert count == 0 diff --git a/tests/execution/test_session_log_integration.py b/tests/execution/test_session_log_integration.py index 5a7780f7..190b21e6 100644 --- a/tests/execution/test_session_log_integration.py +++ b/tests/execution/test_session_log_integration.py @@ -21,7 +21,7 @@ async def test_full_tracing_pipeline_writes_distinct_timestamps(tmp_path): from autoskillit.execution.linux_tracing import start_linux_tracing from autoskillit.execution.session_log import flush_session_log - config = LinuxTracingConfig(proc_interval=0.05) + config = LinuxTracingConfig(proc_interval=0.05, tmpfs_path=str(tmp_path)) start_ts = datetime.now(UTC).isoformat() start_mono = time.monotonic() async with anyio.create_task_group() as tg: diff --git a/tests/infra/test_schema_version_convention.py b/tests/infra/test_schema_version_convention.py index 5e510bc0..d07e7247 100644 --- a/tests/infra/test_schema_version_convention.py +++ b/tests/infra/test_schema_version_convention.py @@ -111,9 +111,9 @@ def _is_yaml_dump(node: ast.expr) -> bool: # core/io.py — write_versioned_json itself (the blessed helper) uses atomic_write+json.dumps ("src/autoskillit/core/io.py", 118), # session_log.py — summary dict, token_usage list, audit_log list - ("src/autoskillit/execution/session_log.py", 206), - ("src/autoskillit/execution/session_log.py", 219), - ("src/autoskillit/execution/session_log.py", 222), + ("src/autoskillit/execution/session_log.py", 213), + ("src/autoskillit/execution/session_log.py", 226), + ("src/autoskillit/execution/session_log.py", 229), # migration/store.py — failure store dicts ("src/autoskillit/migration/store.py", 54), ("src/autoskillit/migration/store.py", 64), @@ -209,8 +209,8 @@ def test_allowlist_includes_list_payloads_as_documented(self): """List-payload sites are included since the AST scanner can't distinguish return types.""" # These sites write list payloads through function calls but are caught by the scanner list_sites = [ - ("src/autoskillit/execution/session_log.py", 219), - ("src/autoskillit/execution/session_log.py", 222), + ("src/autoskillit/execution/session_log.py", 226), + ("src/autoskillit/execution/session_log.py", 229), ("src/autoskillit/smoke_utils.py", 83), ("src/autoskillit/smoke_utils.py", 138), ]