diff --git a/.github/workflows/e2e.yml b/.github/workflows/e2e.yml index ce9c788876..a35b998ce8 100644 --- a/.github/workflows/e2e.yml +++ b/.github/workflows/e2e.yml @@ -107,7 +107,7 @@ jobs: TEST: ${{ matrix.test }} run: | set -euo pipefail - docker_args=() + docker_args=(--shm-size=3g) if [[ '${{ matrix.kind }}' != 'clean-install' ]]; then docker_args+=(--env MINIMAX_API_KEY --env E2E_SANDBOX_TOKEN) docker_args+=(--volume "$RUNNER_TEMP/e2e/data:/home/autoskillit/.local/share/autoskillit") diff --git a/docs/developer/diagnostics.md b/docs/developer/diagnostics.md index 2d373df397..3609f46a28 100644 --- a/docs/developer/diagnostics.md +++ b/docs/developer/diagnostics.md @@ -67,6 +67,68 @@ Logs are stored in a **global** directory (not per-project), so they persist acr `summary.json` contains: `session_id`, `dir_name`, `pid`, `cwd`, `evidence_worktree_path`, `evidence_worktree_source`, `evidence_worktree_detail`, `skill_command`, `success`, `subtype`, `exit_code`, `start_ts`, `snapshot_count`, `anomaly_count`, `peak_rss_kb`, `peak_oom_score`, `peak_fd_ratio`, `session_type`, `session_errors`, `child_outcomes` (see [Child Terminal Reasons](#child-terminal-reasons)). +### Rollout declarations + +`summary.json.rollout` and `sessions.jsonl.rollout` store the same optional +declaration. Configure it in `.autoskillit/config.yaml`: + +```yaml +rollout: + active: + change_id: "prompt-cost-study" + effective_at: "2026-11-01T18:00:00Z" + scope: + skills: ["make-plan"] + backends: ["claude-code", "codex"] + roles: ["planner"] + expected_effect: "Reduce planning input tokens while preserving requirement coverage." +``` + +The stored object has these same fields, with `effective_at` normalized to UTC +ISO-8601 (`2026-11-01T18:00:00+00:00` here). An active declaration must be complete; +`change_id` and `expected_effect` must be nonblank strings. Scope has exactly the +`skills`, `backends`, and `roles` lists of nonblank strings; empty lists are valid. +Scope describes the change and does not filter launches. All admitted Claude Code +and Codex headless runs carry the active marker, including launched failures. +Missing `rollout`, absent/null `active`, or clearing `active` means inactive. +New inactive records store null; older rows without the field remain unassigned. + +Capture occurs after final launch admission, immediately before the first runner +call. The marker and logical start remain fixed through the same-run nudge and +terminal logging even if configuration changes. Summary `start_ts`, index +`timestamp`, and trace enrollment `logical_run_start_ts` share that instant; +operation timing, nudge-turn timing, and enrollment `enrolled_at` remain separate. +Enrollment schema 3 preserves the marker/start for crash recovery. Older sidecars +recover without a marker and retain the historical trace-mtime fallback. + +Declare the prediction and stable change id **before** cutover, launch marked +baseline work, apply the change at the stated UTC boundary, and retain the same +declaration for after runs. Clear `active` when observation ends. A revised +prediction, scope, or cutover requires a new `change_id`; stability across +independent runs is the operator's responsibility. A pre-cutover marked row is +evidence that its prediction was recorded beforehand. A current config or first +after row alone cannot prove that timing; file mtimes do not provide declaration +history. Keeping the prediction allows comparison with observations without +replacing it with an explanation developed after seeing outcomes. + +Use the stored run start and cutover as parsed timezone-aware instants, preserving +fractional seconds. The canonical comparison is: + +```python +started_at = datetime.fromisoformat(row["timestamp"]) +cutover_at = datetime.fromisoformat(rollout["effective_at"]) +if started_at < cutover_at: + cohort = "before" +elif started_at >= cutover_at: + cohort = "after" +``` + +A before/after delta is a prompt to investigate, not proof that the change caused it. +Work difficulty, repository state, and model versions can differ across cohorts. +See the [streaming retained-index query](../operations/observability.md#rollout-cohorts) +for validation and bounded aggregation. Never infer marker membership from hashes, +repository revisions, mtimes, or guessed dates. + ### Execution-candidate manifests Before evaluating a primary binding or configured alternative, `run_skill` @@ -178,7 +240,7 @@ Completed sessions with per-request evidence publish `turn_usage.jsonl` before version-4 token descriptor carries `turn_usage_file`, `turn_usage_count`, and `turn_usage_schema_version`. The file reference is `null` with count zero when no series was observed. The JSONL row schema is version 2; the retained -`sessions.jsonl` index uses schema version 16. +`sessions.jsonl` index uses schema version 17. Each token-bearing row carries non-empty `backend` and `provider_used`, plus nullable source `message_id`, `request_id`, `timestamp`, and observed `model`. diff --git a/docs/operations/observability.md b/docs/operations/observability.md index 9f2561d09e..a8cad35f2e 100644 --- a/docs/operations/observability.md +++ b/docs/operations/observability.md @@ -58,7 +58,7 @@ a named normalization contract. A ratio with an `unknown` required measure renders `unknown`; `unavailable` and `not_applicable` scopes do not contribute to its denominator, and an observed zero denominator renders `—`. -Token descriptor schema v4, turn sidecar schema v2, session index schema v14, +Token descriptor schema v4, turn sidecar schema v2, session index schema v17, fleet campaign state v13, and campaign summary v2 retain these records. Legacy numeric zero is decoded as `unknown` because older formats did not prove an observation. `context_window_tokens` is positive capacity metadata, not a @@ -140,6 +140,106 @@ rows by `dir_name`. Archive-only history excludes the current live survivor window. The archive retains index metadata, not pruned session artifact directories, and operational consumers continue to read only `sessions.jsonl`. +## Rollout cohorts + +Declare an evaluated change in `.autoskillit/config.yaml` before applying it: + +```yaml +rollout: + active: + change_id: "prompt-cost-study" + effective_at: "2026-11-01T18:00:00Z" + scope: + skills: ["make-plan"] + backends: ["claude-code", "codex"] + roles: ["planner"] + expected_effect: "Reduce planning input tokens while preserving requirement coverage." +``` + +`change_id` identifies an unchanged declaration across the observation window. +`effective_at` is the intentional UTC cutover, not a date inferred from code or +configuration history. `scope` names affected skills, backends, and roles in a +queryable form; its lists describe the change and never govern admission or marker +attachment. `expected_effect` preserves the required, non-null, nonblank prediction +with every run so readers can compare intended effects with subsequent observations. +Empty scope lists are valid. Missing/null `active` is inactive; incomplete or +otherwise invalid active declarations are rejected before generic string coercion. + +Declare the prediction and stable id before cutover, launch marked baseline work, +apply the change at the stated UTC boundary, and keep the same declaration active +while observing after runs. Clear `active` when the observation period ends. Use a +new `change_id` for a revised prediction, scope, or cutover. Code validates each +declaration and preserves each admitted run's snapshot; operators maintain stability +across independent runs. Recording the prediction beforehand distinguishes intended +effects from explanations developed after seeing outcomes. A pre-cutover marked +row supplies that evidence; a current configuration or first after row alone does +not establish preregistration. + +Summary and index records store the same nested `rollout` object with normalized +UTC `effective_at`. Capture happens immediately before the first admitted launch; +all headless runs then carry it, including failures and runs outside the descriptive +scope. Same-run nudges and recovery keep that snapshot and logical start despite +later configuration changes. The index `timestamp` and summary `start_ts` identify +the admitted logical run. Operation, nudge-turn, and trace-enrollment times have +different purposes. A subsequent independent run captures the current declaration. +New unmarked rows contain null; older absent fields never imply cohort membership. + +This query counts one explicit change id while streaming the live retained index. +Set `index_path` to `linux_tracing.log_dir / "sessions.jsonl"` if overriding the log +root. It skips reader non-row entries and absent/null markers, and raises a visible +error for malformed selected declarations or timestamps. It covers retained rows +and excludes archived history; memory and printed output do not grow with row count. + +```python +from datetime import datetime, timedelta + +from autoskillit.core import default_log_dir +from autoskillit.execution.session_log.session_index import iter_tolerant_session_index_lines + +index_path = default_log_dir() / "sessions.jsonl" +change_id = "prompt-cost-study" +counts = {"before": 0, "after": 0} +for end_offset, row in iter_tolerant_session_index_lines(index_path): + if row is None: + continue + rollout = row.get("rollout") + if rollout is None: + continue + if not isinstance(rollout, dict): + raise ValueError(f"Malformed rollout at byte {end_offset}") + if rollout["change_id"] != change_id: + continue + if set(rollout) != {"change_id", "effective_at", "scope", "expected_effect"}: + raise ValueError(f"Incomplete rollout at byte {end_offset}") + prediction = rollout["expected_effect"] + scope = rollout["scope"] + if not isinstance(prediction, str) or not prediction.strip(): + raise ValueError(f"Invalid prediction at byte {end_offset}") + if not isinstance(scope, dict) or set(scope) != {"skills", "backends", "roles"}: + raise ValueError(f"Invalid scope at byte {end_offset}") + if any( + not isinstance(values, list) + or any(not isinstance(value, str) or not value.strip() for value in values) + for values in scope.values() + ): + raise ValueError(f"Invalid scope values at byte {end_offset}") + started_at = datetime.fromisoformat(row["timestamp"]) + cutover_at = datetime.fromisoformat(rollout["effective_at"]) + if started_at.utcoffset() is None or cutover_at.utcoffset() != timedelta(0): + raise ValueError(f"Expected aware run start and UTC cutover at byte {end_offset}") + if started_at < cutover_at: + counts["before"] += 1 + elif started_at >= cutover_at: + counts["after"] += 1 +print(counts) +``` + +A before/after delta is a prompt to investigate, not proof that the change caused it. +Work difficulty, repository state, and model versions can differ between cohorts. +Compare parsed instants, including fractional seconds: equality is after cutover. +Do not manufacture cohorts from content hashes, repository revisions, file mtimes, +or guessed dates. Existing aggregate trends do not consume rollout markers. + ## Native headless OTLP capture Every headless executor invocation starts one best-effort OTLP/HTTP-JSON sink diff --git a/src/autoskillit/config/__init__.py b/src/autoskillit/config/__init__.py index 734d847d24..79834a006a 100644 --- a/src/autoskillit/config/__init__.py +++ b/src/autoskillit/config/__init__.py @@ -55,6 +55,7 @@ ResetWorkspaceConfig, RetiredConfigKeyDef, ReviewConfig, + RolloutConfig, RunSkillConfig, SafetyConfig, SkillsConfig, @@ -104,6 +105,7 @@ "RetiredConfigKeyDef", "ReviewConfig", "RunSkillConfig", + "RolloutConfig", "SafetyConfig", "SkillsConfig", "SubsetsConfig", diff --git a/src/autoskillit/config/_automation_config.py b/src/autoskillit/config/_automation_config.py index 149e067eae..33a9835b31 100644 --- a/src/autoskillit/config/_automation_config.py +++ b/src/autoskillit/config/_automation_config.py @@ -44,6 +44,7 @@ from autoskillit.config._dataclasses_execution import ( CodexRuntimeConfig, QuotaGuardConfig, + RolloutConfig, RunSkillConfig, ) from autoskillit.config._dataclasses_fleet import FleetConfig, ProcessTetherConfig @@ -181,6 +182,7 @@ class AutomationConfig: safety: SafetyConfig = field(default_factory=SafetyConfig) read_db: ReadDbConfig = field(default_factory=ReadDbConfig) run_skill: RunSkillConfig = field(default_factory=RunSkillConfig) + rollout: RolloutConfig = field(default_factory=RolloutConfig) model: CoreRunConfig = field(default_factory=CoreRunConfig) worktree_setup: WorktreeSetupConfig = field(default_factory=WorktreeSetupConfig) migration: MigrationConfig = field(default_factory=MigrationConfig) diff --git a/src/autoskillit/config/_coercion.py b/src/autoskillit/config/_coercion.py index 9797322f70..76978c9226 100644 --- a/src/autoskillit/config/_coercion.py +++ b/src/autoskillit/config/_coercion.py @@ -19,9 +19,12 @@ import dataclasses import types -from collections.abc import Callable +from collections.abc import Callable, Mapping +from datetime import UTC, datetime, timedelta from typing import TYPE_CHECKING, Any, TypeVar, Union, get_args, get_origin, get_type_hints +import regex as re + from autoskillit.config._config_loader import ( _build_packs_config, _build_subsets_config, @@ -29,6 +32,7 @@ ) from autoskillit.config._dataclasses_errors import ConfigSchemaError from autoskillit.config._dataclasses_test_gating import _COMMAND_UNSET +from autoskillit.core import RolloutMarker if TYPE_CHECKING: from _typeshed import DataclassInstance @@ -190,10 +194,64 @@ def _coerce_value(value: Any, target_type: type, context: str) -> Any: } +def _rollout_cutover_utc(cutover: Any) -> str: + if not isinstance(cutover, str) or re.search(r"(?:Z|[+-]00:00)$", cutover) is None: + raise ConfigSchemaError( + "rollout.active.effective_at must be an explicit UTC ISO-8601 timestamp" + ) + try: + instant = datetime.fromisoformat(cutover.replace("Z", "+00:00")) + except ValueError as exc: + raise ConfigSchemaError( + "rollout.active.effective_at must be an explicit UTC ISO-8601 timestamp" + ) from exc + if instant.utcoffset() != timedelta(0): + raise ConfigSchemaError("rollout.active.effective_at must be UTC") + return instant.astimezone(UTC).isoformat() + + +def _build_rollout_active( + section: dict[str, Any], defaults: dict[str, Any] +) -> RolloutMarker | None: + raw = section.get("active", defaults.get("active")) + if raw is None: + return None + keys = {"change_id", "effective_at", "scope", "expected_effect"} + if not isinstance(raw, Mapping) or set(raw) != keys: + raise ConfigSchemaError( + "rollout.active must contain exactly change_id, effective_at, scope, expected_effect" + ) + for name in ("change_id", "expected_effect"): + if not isinstance(raw[name], str) or not raw[name].strip(): + raise ConfigSchemaError(f"rollout.active.{name} must be a nonblank string") + effective_at = _rollout_cutover_utc(raw["effective_at"]) + scope = raw["scope"] + if not isinstance(scope, Mapping) or set(scope) != {"skills", "backends", "roles"}: + raise ConfigSchemaError( + "rollout.active.scope must contain exactly skills, backends, roles" + ) + copied_scope: dict[str, list[str]] = {} + for axis, values in scope.items(): + if not isinstance(values, list) or any( + not isinstance(value, str) or not value.strip() for value in values + ): + raise ConfigSchemaError( + f"rollout.active.scope.{axis} must be a list of nonblank strings" + ) + copied_scope[axis] = list(values) + return RolloutMarker( + change_id=raw["change_id"], + effective_at=effective_at, + scope=copied_scope, + expected_effect=raw["expected_effect"], + ) + + # Custom field builders that bypass _coerce_value. # Signature: (section_dict, defaults_dict) -> coerced_value # The override is responsible for its own key lookup from section_dict. _FIELD_OVERRIDES: dict[tuple[str, str], Callable[[dict[str, Any], dict[str, Any]], Any]] = { + ("rollout", "active"): _build_rollout_active, # YAML key "default" with None-means-unset semantic. # ``defs.get(...)`` (no explicit fallback) is structurally defensive — if the # dataclass field is ever removed without updating this lambda, the .get() diff --git a/src/autoskillit/config/_config_dataclasses.py b/src/autoskillit/config/_config_dataclasses.py index a0de1cc138..e1723d1f03 100644 --- a/src/autoskillit/config/_config_dataclasses.py +++ b/src/autoskillit/config/_config_dataclasses.py @@ -59,6 +59,9 @@ from autoskillit.config._dataclasses_execution import ( QuotaGuardConfig as QuotaGuardConfig, ) +from autoskillit.config._dataclasses_execution import ( + RolloutConfig as RolloutConfig, +) from autoskillit.config._dataclasses_execution import ( RunSkillConfig as RunSkillConfig, ) diff --git a/src/autoskillit/config/_dataclasses_execution.py b/src/autoskillit/config/_dataclasses_execution.py index d2cd83d168..941152874d 100644 --- a/src/autoskillit/config/_dataclasses_execution.py +++ b/src/autoskillit/config/_dataclasses_execution.py @@ -14,7 +14,12 @@ from dataclasses import dataclass, field from typing import ClassVar, Literal, cast -from autoskillit.core import CodexRuntimeSpec, OutputFormat +from autoskillit.core import CodexRuntimeSpec, OutputFormat, RolloutMarker + + +@dataclass +class RolloutConfig: + active: RolloutMarker | None = None @dataclass diff --git a/src/autoskillit/config/settings.py b/src/autoskillit/config/settings.py index 4ae954cb80..69c906d05f 100644 --- a/src/autoskillit/config/settings.py +++ b/src/autoskillit/config/settings.py @@ -149,6 +149,9 @@ from autoskillit.config._config_dataclasses import ( ReviewConfig as ReviewConfig, ) +from autoskillit.config._config_dataclasses import ( + RolloutConfig as RolloutConfig, +) from autoskillit.config._config_dataclasses import ( RunSkillConfig as RunSkillConfig, ) @@ -271,6 +274,7 @@ "RetiredConfigKeyDef", "ReviewConfig", "RunSkillConfig", + "RolloutConfig", "SafetyConfig", "SkillsConfig", "SubsetsConfig", diff --git a/src/autoskillit/core/__init__.pyi b/src/autoskillit/core/__init__.pyi index 25ccba53eb..609d4e8a9b 100644 --- a/src/autoskillit/core/__init__.pyi +++ b/src/autoskillit/core/__init__.pyi @@ -1266,6 +1266,7 @@ from .types import ReviewOperationState as ReviewOperationState from .types import ReviewReconciliationResult as ReviewReconciliationResult from .types import ReviewResponseClass as ReviewResponseClass from .types import RollbackAdmissionEvent as RollbackAdmissionEvent +from .types import RolloutMarker as RolloutMarker from .types import RolloverEpochEvent as RolloverEpochEvent from .types import RunSkillCompletionAuthority as RunSkillCompletionAuthority from .types import RuntimeAdmission as RuntimeAdmission diff --git a/src/autoskillit/core/types/execution/_type_subprocess.py b/src/autoskillit/core/types/execution/_type_subprocess.py index 392b52680e..abcc9ab1b8 100644 --- a/src/autoskillit/core/types/execution/_type_subprocess.py +++ b/src/autoskillit/core/types/execution/_type_subprocess.py @@ -12,6 +12,7 @@ from typing import Any, Protocol, runtime_checkable from ..foundation._type_enums import ChannelConfirmation, KillReason, TerminationReason +from ..results._type_results_records import RolloutMarker from ._type_inspector import InspectorCallback, InspectorVerdict __all__ = [ @@ -258,6 +259,8 @@ def __call__( completion_drain_timeout: float = 5.0, natural_exit_grace_seconds: float = 3.0, linux_tracing_config: Any | None = None, + rollout: RolloutMarker | None = None, + logical_run_start_ts: str = "", idle_output_timeout: float | None = None, max_suppression_seconds: float | None = None, on_process_spawned: Callable[[int, int], None] | None = None, diff --git a/src/autoskillit/core/types/results/_type_results.py b/src/autoskillit/core/types/results/_type_results.py index aa4bf37df5..277bbd6a97 100644 --- a/src/autoskillit/core/types/results/_type_results.py +++ b/src/autoskillit/core/types/results/_type_results.py @@ -51,6 +51,7 @@ RecipePathValidationInputError, RecipePathValidationReport, RecipePathValidationResult, + RolloutMarker, SessionIndexEntry, SkillUnavailabilityPayload, SkillUnavailabilityRecord, @@ -126,6 +127,7 @@ "ModelTotalEntry", "TokenUsageFileEntry", "SessionIndexEntry", + "RolloutMarker", "SESSION_INDEX_SCHEMA_VERSION", "TOKEN_USAGE_SCHEMA_VERSION", "TURN_USAGE_SCHEMA_VERSION", diff --git a/src/autoskillit/core/types/results/_type_results_records.py b/src/autoskillit/core/types/results/_type_results_records.py index ec8f007a90..c6afeec2fd 100644 --- a/src/autoskillit/core/types/results/_type_results_records.py +++ b/src/autoskillit/core/types/results/_type_results_records.py @@ -45,13 +45,14 @@ "EvidenceWorktreeSource", "ValidatedAddDir", "ValidatedWorktreePath", + "RolloutMarker", "TokenUsageFileEntry", "SessionIndexEntry", ] T = TypeVar("T") -SESSION_INDEX_SCHEMA_VERSION: Final[int] = 16 +SESSION_INDEX_SCHEMA_VERSION: Final[int] = 17 TOKEN_USAGE_SCHEMA_VERSION: Final[int] = 4 TURN_USAGE_SCHEMA_VERSION: Final[int] = 2 @@ -361,12 +362,22 @@ class TokenUsageFileEntry(TypedDict): schema_version: int +class RolloutMarker(TypedDict): + """Run declaration; scope axes are skills, backends, and roles.""" + + change_id: str + effective_at: str + scope: dict[str, list[str]] + expected_effect: str + + class SessionIndexEntry(TypedDict): """Schema contract for sessions.jsonl entries written by flush_session_log.""" session_id: str dir_name: str timestamp: str + rollout: RolloutMarker | None cwd: str evidence_worktree_path: str evidence_worktree_source: str diff --git a/src/autoskillit/execution/evidence/linux_tracing.py b/src/autoskillit/execution/evidence/linux_tracing.py index 25114067b7..8748efa288 100644 --- a/src/autoskillit/execution/evidence/linux_tracing.py +++ b/src/autoskillit/execution/evidence/linux_tracing.py @@ -32,8 +32,8 @@ import anyio.abc import psutil +from autoskillit.core import RolloutMarker, get_logger from autoskillit.core import fast_dumps as _fast_dumps -from autoskillit.core import get_logger from autoskillit.core import is_pid_zombie as is_pid_zombie from autoskillit.core import read_boot_id as read_boot_id from autoskillit.core import read_starttime_ticks as read_starttime_ticks @@ -199,7 +199,7 @@ def trace_target_from_pid(pid: int) -> TraceTarget: # --------------------------------------------------------------------------- -# TraceEnrollmentRecord: schema version 2 (adds comm field) +# TraceEnrollmentRecord: schema version 3 (adds rollout and logical run start) # --------------------------------------------------------------------------- @@ -213,9 +213,10 @@ class TraceEnrollmentRecord: Schema version 2 adds 'comm' to the enrollment record so that crash recovery can verify the process identity (not just its PID) and reject alien trace files. + Schema version 3 preserves the admitted rollout declaration and logical run start. """ - schema_version: int # 2 for post-#806 records; 1 for pre-fix records + schema_version: int pid: int boot_id: str | None # read_boot_id(); None if /proc unavailable starttime_ticks: int | None # read_starttime_ticks(pid); None if unavailable @@ -224,6 +225,8 @@ class TraceEnrollmentRecord: kitchen_id: str # "" order_id: str # "" comm: str = "" # /proc/{pid}/comm value; "" for pre-fix (schema_version=1) records + rollout: RolloutMarker | None = None + logical_run_start_ts: str = "" def _write_enrollment_atomic(path: Path, record: TraceEnrollmentRecord) -> None: @@ -256,8 +259,15 @@ def read_enrollment(path: Path) -> TraceEnrollmentRecord | None: kitchen_id=data.get("kitchen_id", ""), order_id=data.get("order_id", ""), comm=data.get("comm", ""), # "" for pre-fix schema_version=1 records + rollout=data.get("rollout"), + logical_run_start_ts=data.get("logical_run_start_ts", ""), + ) + except FileNotFoundError: + return None + except (OSError, KeyError, TypeError, ValueError) as exc: + logger.warning( + "Failed to read enrollment sidecar %s (%s): %s", path, type(exc).__name__, exc ) - except (OSError, KeyError, TypeError, ValueError, json.JSONDecodeError): return None @@ -578,6 +588,8 @@ def start_linux_tracing( session_id: str = "", kitchen_id: str = "", order_id: str = "", + rollout: RolloutMarker | None = None, + logical_run_start_ts: str = "", ) -> LinuxTracingHandle | None: """Start Linux tracing if all gates pass. Returns handle or None. @@ -614,11 +626,10 @@ def start_linux_tracing( handle._trace_file = None # Write enrollment sidecar atomically for crash-recovery identity contract - # Schema version 2: includes comm field for alien-file rejection in recovery enrollment_path = tmpfs / f"autoskillit_enrollment_{pid}.json" try: record = TraceEnrollmentRecord( - schema_version=2, + schema_version=3, pid=pid, boot_id=read_boot_id(), starttime_ticks=target.starttime_ticks, @@ -627,6 +638,8 @@ def start_linux_tracing( kitchen_id=kitchen_id, order_id=order_id, comm=target.comm, + rollout=rollout, + logical_run_start_ts=logical_run_start_ts, ) _write_enrollment_atomic(enrollment_path, record) handle._enrollment_path = enrollment_path diff --git a/src/autoskillit/execution/headless/_headless_execute.py b/src/autoskillit/execution/headless/_headless_execute.py index aebe7be4d6..92d3a014b7 100644 --- a/src/autoskillit/execution/headless/_headless_execute.py +++ b/src/autoskillit/execution/headless/_headless_execute.py @@ -7,6 +7,7 @@ import time import traceback from collections.abc import Awaitable, Callable, Mapping, Sequence +from copy import deepcopy from datetime import UTC, datetime, timedelta from pathlib import Path from types import TracebackType @@ -32,6 +33,7 @@ PluginLoadMode, ResolvedLaunchContract, RetryReason, + RolloutMarker, SkillResult, WriteBehaviorSpec, collect_version_snapshot, @@ -191,6 +193,17 @@ async def _execute_claude_headless( _start_ts = datetime.now(UTC).isoformat() _start_mono = time.monotonic() _start_epoch = time.time() + _rollout_snapshot: RolloutMarker | None = None + _logical_run_start_ts: str | None = None + + def capture_rollout_on_launch() -> tuple[RolloutMarker | None, str]: + """Capture marker and UTC start once after admission; reuse across retries and nudges.""" + nonlocal _rollout_snapshot, _logical_run_start_ts + if _logical_run_start_ts is None: + _rollout_snapshot = deepcopy(ctx.config.rollout.active) + _logical_run_start_ts = datetime.now(UTC).isoformat() + return _rollout_snapshot, _logical_run_start_ts + _versions = collect_version_snapshot(_step_backend) # type: ignore[arg-type] _readonly_skill = readonly_skill @@ -346,6 +359,7 @@ def observe_execution_started() -> None: completion_drain_timeout=cfg.completion_drain_timeout, natural_exit_grace_seconds=natural_exit_grace_seconds, linux_tracing_config=linux_tracing_cfg, + capture_rollout_on_launch=capture_rollout_on_launch, idle_output_timeout=base_effective_idle, max_suppression_seconds=cfg.max_suppression_seconds, child_deferral_ceiling=cfg.completion_child_deferral_ceiling_seconds, @@ -506,6 +520,8 @@ def observe_execution_started() -> None: mark_execution_started=observe_execution_started, on_session_id_resolved=capture_resolved_session_id, natural_exit_grace_seconds=natural_exit_grace_seconds, + capture_rollout_on_launch=capture_rollout_on_launch, + linux_tracing_config=linux_tracing_cfg, nudge_timeout=min(60.0, nudge_deadline_remaining), attempt=physical_attempt, ceiling_seconds=ceiling_seconds, @@ -654,21 +670,7 @@ def observe_execution_started() -> None: execution_selection=terminal_selection, ) - if terminal_selection is not None: - from autoskillit.execution.session_log.session_log import ( - write_execution_candidate_manifest, - ) - - try: - write_execution_candidate_manifest( - terminal_selection, - ctx.config.linux_tracing.log_dir, - max_sessions=ctx.config.linux_tracing.max_sessions, - project_dir=str(ctx.project_dir), - build_protected_campaign_ids=ctx.build_protected_campaign_ids, - ) - except Exception: - logger.debug("execution_candidate_manifest_write_failed", exc_info=True) + _terminal.persist_terminal_selection(ctx, terminal_selection) terminal_capture_diagnostic = _diag.capture(managed_lineage_observer) if _diag.should_flush(result, skill_result, step_name, terminal_capture_diagnostic): @@ -692,6 +694,8 @@ def observe_execution_started() -> None: skill_command=skill_command, step_name=step_name, start_ts=_start_ts, + rollout=_rollout_snapshot, + logical_run_start_ts=_logical_run_start_ts, termination_reason=terminal_reason_override, exception_text=terminal_exception_text, versions=_versions, diff --git a/src/autoskillit/execution/headless/_headless_launch.py b/src/autoskillit/execution/headless/_headless_launch.py index d4f7d18316..35409d651d 100644 --- a/src/autoskillit/execution/headless/_headless_launch.py +++ b/src/autoskillit/execution/headless/_headless_launch.py @@ -20,6 +20,7 @@ PluginLoadMode, ResolvedLaunchContract, RetryReason, + RolloutMarker, SkillContractView, SkillResult, StreamParser, @@ -64,21 +65,6 @@ def _report_plugin_binding_close_failure(error: BaseException, _: BaseException) logger.warning("plugin_launch_binding_close_failed", primary_error=repr(error), exc_info=True) -def _plugin_launch_binding( - *, - authority: PluginArtifactAuthority | None, - backend: CodingAgentBackend, - load_mode: PluginLoadMode, -) -> AbstractContextManager[PluginLaunchBinding | None]: - """Own one exact artifact binding for the complete child attempt.""" - return plugin_launch_binding_scope( - authority=authority, - backend=backend, - load_mode=load_mode, - on_suppressed_close_error=_report_plugin_binding_close_failure, - ) - - async def _run_headless_attempt( build_spec: _BuildSpec, *, @@ -97,6 +83,7 @@ async def _run_headless_attempt( completion_drain_timeout: float, natural_exit_grace_seconds: float, linux_tracing_config: Any, + capture_rollout_on_launch: Callable[[], tuple[RolloutMarker | None, str]], idle_output_timeout: float | None, max_suppression_seconds: float, child_deferral_ceiling: float, @@ -126,10 +113,11 @@ async def _run_headless_attempt( binding_scope: AbstractContextManager[PluginLaunchBinding | None] = ( nullcontext(retained_binding) if retained_binding is not None - else _plugin_launch_binding( + else plugin_launch_binding_scope( authority=plugin_authority, backend=backend, load_mode=plugin_load_mode, + on_suppressed_close_error=_report_plugin_binding_close_failure, ) ) with binding_scope as binding: @@ -187,6 +175,7 @@ async def _run_headless_attempt( on_spec_built(spec) if mark_execution_started is not None: mark_execution_started() + rollout, logical_run_start_ts = capture_rollout_on_launch() result = await runner( list(spec.cmd), cwd=Path(spec.cwd), @@ -201,6 +190,8 @@ async def _run_headless_attempt( completion_drain_timeout=completion_drain_timeout, natural_exit_grace_seconds=natural_exit_grace_seconds, linux_tracing_config=linux_tracing_config, + rollout=rollout, + logical_run_start_ts=logical_run_start_ts, idle_output_timeout=idle_output_timeout, max_suppression_seconds=max_suppression_seconds, child_deferral_ceiling=child_deferral_ceiling, @@ -234,6 +225,8 @@ async def _attempt_contract_nudge( cwd: str, runner: SubprocessRunner, *, + capture_rollout_on_launch: Callable[[], tuple[RolloutMarker | None, str]], + linux_tracing_config: Any | None = None, backend: CodingAgentBackend | None = None, result_parser: ResultParser | None = None, provider_extras: Mapping[str, str] | None = None, @@ -326,10 +319,11 @@ async def _attempt_contract_nudge( nudge_binding_scope: AbstractContextManager[PluginLaunchBinding | None] = ( nullcontext(retained_binding) if retained_binding is not None - else _plugin_launch_binding( + else plugin_launch_binding_scope( authority=plugin_authority, backend=backend, load_mode=plugin_load_mode, + on_suppressed_close_error=_report_plugin_binding_close_failure, ) ) try: @@ -430,6 +424,7 @@ def build_nudge_spec( nudge_start_ts = datetime.now(UTC).isoformat() if mark_execution_started is not None: mark_execution_started() + rollout, logical_run_start_ts = capture_rollout_on_launch() nudge_result = await runner( list(spec.cmd), cwd=Path(spec.cwd), @@ -443,6 +438,9 @@ def build_nudge_spec( on_process_reaped=handle.record_reaped if handle is not None else None, ceiling_seconds=ceiling_seconds, natural_exit_grace_seconds=natural_exit_grace_seconds, + rollout=rollout, + logical_run_start_ts=logical_run_start_ts, + linux_tracing_config=linux_tracing_config, line_driver=backend.line_driver(spec), ) nudge_end_ts = datetime.now(UTC).isoformat() diff --git a/src/autoskillit/execution/headless/_headless_terminal.py b/src/autoskillit/execution/headless/_headless_terminal.py index 49e0c42a0f..df19d770b0 100644 --- a/src/autoskillit/execution/headless/_headless_terminal.py +++ b/src/autoskillit/execution/headless/_headless_terminal.py @@ -5,7 +5,7 @@ import dataclasses import time from collections.abc import Callable -from typing import Any +from typing import TYPE_CHECKING, Any from autoskillit.core import ( CodingAgentBackend, @@ -20,12 +20,37 @@ RecipeIdentity, ResolvedLaunchContract, RetryReason, + RolloutMarker, SessionTelemetry, SessionType, SkillResult, SubprocessResult, + get_logger, ) +if TYPE_CHECKING: + from autoskillit.pipeline.context import ToolContext + +logger = get_logger(__name__) + + +def persist_terminal_selection(ctx: ToolContext, selection: ExecutionSelection | None) -> None: + """Publish the final candidate manifest without interrupting session logging.""" + if selection is None: + return + from autoskillit.execution.session_log.session_log import write_execution_candidate_manifest + + try: + write_execution_candidate_manifest( + selection, + ctx.config.linux_tracing.log_dir, + max_sessions=ctx.config.linux_tracing.max_sessions, + project_dir=str(ctx.project_dir), + build_protected_campaign_ids=ctx.build_protected_campaign_ids, + ) + except Exception: + logger.debug("execution_candidate_manifest_write_failed", exc_info=True) + def reconcile_token_evidence( skill_result: SkillResult, @@ -290,6 +315,8 @@ def build_terminal_flush_kwargs( skill_command: str, step_name: str, start_ts: str, + rollout: RolloutMarker | None, + logical_run_start_ts: str | None, termination_reason: str, exception_text: str, versions: dict[str, Any], @@ -332,7 +359,8 @@ def build_terminal_flush_kwargs( "is_error": skill_result.is_error, "subtype": skill_result.subtype, "exit_code": skill_result.exit_code, - "start_ts": result.start_ts if result is not None else start_ts, + "start_ts": logical_run_start_ts if logical_run_start_ts is not None else start_ts, + "rollout": rollout, "proc_snapshots": result.proc_snapshots if result is not None else None, "termination_reason": ( result.termination.value if result is not None else termination_reason diff --git a/src/autoskillit/execution/process/__init__.py b/src/autoskillit/execution/process/__init__.py index 57959232d1..115be7083e 100644 --- a/src/autoskillit/execution/process/__init__.py +++ b/src/autoskillit/execution/process/__init__.py @@ -24,6 +24,7 @@ from autoskillit.core import ( ChannelBStatus, ChannelConfirmation, + RolloutMarker, SessionEvent, SubprocessResult, TerminationReason, @@ -249,6 +250,8 @@ async def run_managed_async( completion_drain_timeout: float = 5.0, natural_exit_grace_seconds: float = 3.0, linux_tracing_config: LinuxTracingConfig | None = None, + rollout: RolloutMarker | None = None, + logical_run_start_ts: str = "", idle_output_timeout: float | None = None, max_suppression_seconds: float | None = None, on_process_spawned: Callable[[int, int], None] | None = None, @@ -542,6 +545,8 @@ async def run_managed_async( target=_target, config=linux_tracing_config, tg=tg, + rollout=rollout, + logical_run_start_ts=logical_run_start_ts, ) timeout_scope = await _await_race_and_drain( tg, diff --git a/src/autoskillit/execution/process/_lifecycle/runner.py b/src/autoskillit/execution/process/_lifecycle/runner.py index 0a43cc0b04..8e274a3e58 100644 --- a/src/autoskillit/execution/process/_lifecycle/runner.py +++ b/src/autoskillit/execution/process/_lifecycle/runner.py @@ -10,6 +10,7 @@ from autoskillit.core import ( ChannelConfirmation, + RolloutMarker, SubprocessResult, TerminationReason, get_logger, @@ -149,6 +150,8 @@ async def __call__( completion_drain_timeout: float = 5.0, natural_exit_grace_seconds: float = 3.0, linux_tracing_config: LinuxTracingConfig | None = None, + rollout: RolloutMarker | None = None, + logical_run_start_ts: str = "", idle_output_timeout: float | None = None, max_suppression_seconds: float | None = None, on_process_spawned: Callable[[int, int], None] | None = None, @@ -188,6 +191,8 @@ async def __call__( completion_drain_timeout=completion_drain_timeout, natural_exit_grace_seconds=natural_exit_grace_seconds, linux_tracing_config=linux_tracing_config, + rollout=rollout, + logical_run_start_ts=logical_run_start_ts, idle_output_timeout=idle_output_timeout, max_suppression_seconds=max_suppression_seconds, on_process_spawned=on_process_spawned, diff --git a/src/autoskillit/execution/recording/recording.py b/src/autoskillit/execution/recording/recording.py index a22e9615fe..d4f7519ae1 100644 --- a/src/autoskillit/execution/recording/recording.py +++ b/src/autoskillit/execution/recording/recording.py @@ -18,6 +18,7 @@ CLAUDE_CODE_CAPABILITIES, BackendCapabilities, InspectorCallback, + RolloutMarker, SubprocessResult, SubprocessRunner, TerminationReason, @@ -136,6 +137,8 @@ async def __call__( completion_drain_timeout: float = 5.0, natural_exit_grace_seconds: float = 3.0, linux_tracing_config: Any | None = None, + rollout: RolloutMarker | None = None, + logical_run_start_ts: str = "", idle_output_timeout: float | None = None, max_suppression_seconds: float | None = None, on_process_spawned: Callable[[int, int], None] | None = None, @@ -193,6 +196,8 @@ async def __call__( completion_drain_timeout=completion_drain_timeout, natural_exit_grace_seconds=natural_exit_grace_seconds, linux_tracing_config=linux_tracing_config, + rollout=rollout, + logical_run_start_ts=logical_run_start_ts, idle_output_timeout=idle_output_timeout, max_suppression_seconds=max_suppression_seconds, on_process_spawned=on_process_spawned, @@ -231,6 +236,8 @@ async def __call__( completion_drain_timeout=completion_drain_timeout, natural_exit_grace_seconds=natural_exit_grace_seconds, linux_tracing_config=linux_tracing_config, + rollout=rollout, + logical_run_start_ts=logical_run_start_ts, idle_output_timeout=idle_output_timeout, max_suppression_seconds=max_suppression_seconds, on_process_spawned=on_process_spawned, @@ -278,6 +285,8 @@ async def __call__( completion_drain_timeout=completion_drain_timeout, natural_exit_grace_seconds=natural_exit_grace_seconds, linux_tracing_config=linux_tracing_config, + rollout=rollout, + logical_run_start_ts=logical_run_start_ts, idle_output_timeout=idle_output_timeout, max_suppression_seconds=max_suppression_seconds, on_process_spawned=on_process_spawned, @@ -387,6 +396,8 @@ async def _record_non_pty_session( completion_drain_timeout: float, natural_exit_grace_seconds: float, linux_tracing_config: Any | None, + rollout: RolloutMarker | None, + logical_run_start_ts: str, idle_output_timeout: float | None, max_suppression_seconds: float | None, on_process_spawned: Callable[[int, int], None] | None, @@ -424,6 +435,8 @@ async def _record_non_pty_session( completion_drain_timeout=completion_drain_timeout, natural_exit_grace_seconds=natural_exit_grace_seconds, linux_tracing_config=linux_tracing_config, + rollout=rollout, + logical_run_start_ts=logical_run_start_ts, idle_output_timeout=idle_output_timeout, max_suppression_seconds=max_suppression_seconds, on_process_spawned=on_process_spawned, @@ -542,6 +555,8 @@ async def __call__( completion_drain_timeout: float = 5.0, natural_exit_grace_seconds: float = 3.0, linux_tracing_config: Any | None = None, + rollout: RolloutMarker | None = None, + logical_run_start_ts: str = "", idle_output_timeout: float | None = None, max_suppression_seconds: float | None = None, on_process_spawned: Callable[[int, int], None] | None = None, @@ -580,6 +595,8 @@ async def __call__( ceiling_seconds, systemd_scope_enabled, natural_exit_grace_seconds, + rollout, + logical_run_start_ts, ) step_name = (env or {}).get(SCENARIO_STEP_NAME_ENV, "") diff --git a/src/autoskillit/execution/session_log/_session_log_recovery.py b/src/autoskillit/execution/session_log/_session_log_recovery.py index 1e041112ef..12e62808e7 100644 --- a/src/autoskillit/execution/session_log/_session_log_recovery.py +++ b/src/autoskillit/execution/session_log/_session_log_recovery.py @@ -131,6 +131,7 @@ def _finalize_crashed_trace( build_protected_campaign_ids: CampaignProtector | None, pid: int, snapshots: list[dict[str, object]], + enrollment: TraceEnrollmentRecord, ) -> bool: """Write one decoded crash trace into the durable session log.""" try: @@ -160,7 +161,8 @@ def _finalize_crashed_trace( is_error=True, subtype="crashed", exit_code=-1, - start_ts=mtime_ts, + start_ts=enrollment.logical_run_start_ts or mtime_ts, + rollout=enrollment.rollout, proc_snapshots=snapshots if snapshots else None, termination_reason="CRASHED", provider_outcome=ProviderOutcome.none_used(), @@ -245,6 +247,7 @@ def recover_crashed_sessions( build_protected_campaign_ids=build_protected_campaign_ids, pid=pid, snapshots=snapshots, + enrollment=enrollment, ): _delete_trace_pair(trace_file, enrollment_path) count += 1 diff --git a/src/autoskillit/execution/session_log/session_log.py b/src/autoskillit/execution/session_log/session_log.py index 0652034458..ee46fc73a8 100644 --- a/src/autoskillit/execution/session_log/session_log.py +++ b/src/autoskillit/execution/session_log/session_log.py @@ -31,6 +31,7 @@ EvidenceWorktree, EvidenceWorktreeSource, ModelIdentity, + RolloutMarker, SessionType, atomic_write, default_log_dir, @@ -158,6 +159,7 @@ def flush_session_log( subtype: str, exit_code: int, start_ts: str, + rollout: RolloutMarker | None = None, proc_snapshots: list[dict[str, object]] | None, end_ts: str = "", elapsed_seconds: float | None = None, @@ -413,6 +415,7 @@ def flush_session_log( ) summary = { + "rollout": rollout, "session_id": session_id, "dir_name": dir_name, "pid": pid, @@ -582,6 +585,7 @@ def flush_session_log( refresh_summary_child_outcomes(summary_path, list(telemetry.child_outcomes)) index_entry = { + "rollout": rollout, "session_id": session_id, "dir_name": dir_name, "timestamp": start_ts, diff --git a/tests/_helpers.py b/tests/_helpers.py index a75b0fddd7..43e5e6f5c8 100644 --- a/tests/_helpers.py +++ b/tests/_helpers.py @@ -20,6 +20,7 @@ from autoskillit.core import ( InstructionExtractionMode, OrchestratorSurfaceDef, + RolloutMarker, ToolParamRole, get_tool_def, ) @@ -38,6 +39,19 @@ INVALID_PATH_INPUTS: tuple[object, ...] = (None, 0, b"", object(), MagicMock()) +def make_rollout_marker(*, backends: tuple[str, ...] = ("claude-code", "codex")) -> RolloutMarker: + return { + "change_id": "prompt-cost-study", + "effective_at": "2026-11-01T18:00:00+00:00", + "scope": { + "skills": ["make-plan"], + "backends": list(backends), + "roles": ["planner"], + }, + "expected_effect": "Reduce planning input tokens while preserving requirement coverage.", + } + + def load_core_stub_generator() -> ModuleType: """Load a fresh generator module so callers can independently patch its paths.""" script = Path(__file__).resolve().parent.parent / "scripts" / "generate_core_stub.py" diff --git a/tests/_retention_surface.py b/tests/_retention_surface.py index b0fc2bcc76..47a8d6b96b 100644 --- a/tests/_retention_surface.py +++ b/tests/_retention_surface.py @@ -1435,7 +1435,7 @@ def assert_second_pass_is_quiet( # the #4623 pass's own child_outcomes import to inside the try block (issue # #4672 decomposition — module-level would circularly import back through # the evidence/ gateway that now wraps this file). - f"{_SR}::L231": _retries_after_input_changes( + f"{_SR}::L233": _retries_after_input_changes( "The configured trace root is absent, so no crash candidate can be discovered yet." ), f"{_SRE}::L45": _retries_after_input_changes( @@ -1474,10 +1474,10 @@ def assert_second_pass_is_quiet( f"{_SRD}::L121": _self_limiting( "An alien-command trace and its enrollment are deleted as a terminal safety disposition." ), - f"{_SRF}::L139": _retries_after_input_changes( + f"{_SRF}::L140": _retries_after_input_changes( "A second stat failure keeps the trace retryable until the filesystem becomes available." ), - f"{_SRF}::L180": _retries_after_input_changes( + f"{_SRF}::L182": _retries_after_input_changes( "Flush or output-index failure retains both files until output infrastructure recovers." ), } diff --git a/tests/_test_filter.py b/tests/_test_filter.py index 5a8cfcb557..8c82ec3619 100644 --- a/tests/_test_filter.py +++ b/tests/_test_filter.py @@ -550,7 +550,7 @@ class SourceMapRejection(enum.StrEnum): "tool_sequence_analysis": frozenset({"core", "execution", "fleet", "server", "cli"}), "_type_checkpoint": frozenset({"core", "execution", "fleet", "server"}), "_type_results": _TYPE_RESULTS_CASCADE, - "_type_results_records": _TYPE_RESULTS_CASCADE, + "_type_results_records": _TYPE_RESULTS_CASCADE | {"config"}, "_type_results_execution": frozenset({"core", "execution", "server", "pipeline"}), "_type_execution_identity": frozenset( {"core", "execution", "server", "cli", "fleet", "smoke_utils"} diff --git a/tests/arch/_core_gateway_snapshot.py b/tests/arch/_core_gateway_snapshot.py index 8ec74d5d3c..03914be0ca 100644 --- a/tests/arch/_core_gateway_snapshot.py +++ b/tests/arch/_core_gateway_snapshot.py @@ -762,6 +762,7 @@ "ReviewResponseClass", "Revocability", "RollbackAdmissionEvent", + "RolloutMarker", "RolloverEpochEvent", "RunSkillCompletionAuthority", "RuntimeAdmission", diff --git a/tests/arch/test_execution_source_split.py b/tests/arch/test_execution_source_split.py index 68b1595ca8..781cbc1252 100644 --- a/tests/arch/test_execution_source_split.py +++ b/tests/arch/test_execution_source_split.py @@ -1,5 +1,6 @@ """Arch guards for execution layer source splits (P8-F1, P8-F3, P8-F4 audit fixes).""" +import ast from pathlib import Path import pytest @@ -85,12 +86,22 @@ def test_headless_facade_does_not_define_recovery_functions(): def test_plugin_bound_launch_lifecycle_is_centralized(): launch = (SRC_EXECUTION / "headless" / "_headless_launch.py").read_text() + functions = { + node.name: node + for node in ast.parse(launch).body + if isinstance(node, ast.AsyncFunctionDef) + } for fn in ( - "_plugin_launch_binding", "_run_headless_attempt", "_attempt_contract_nudge", ): - assert f"def {fn}" in launch, f"{fn} must live in _headless_launch.py" + assert fn in functions, f"{fn} must live in _headless_launch.py" + assert any( + isinstance(node, ast.Call) + and isinstance(node.func, ast.Name) + and node.func.id == "plugin_launch_binding_scope" + for node in ast.walk(functions[fn]) + ), f"{fn} must own its plugin binding scope" def test_headless_facade_does_not_define_path_token_functions(): diff --git a/tests/arch/test_import_linter_contracts.py b/tests/arch/test_import_linter_contracts.py index f30a2a3303..6a39d9c320 100644 --- a/tests/arch/test_import_linter_contracts.py +++ b/tests/arch/test_import_linter_contracts.py @@ -42,6 +42,7 @@ "execution/headless/_headless_outcome.py": frozenset({"recipe"}), "execution/headless/_headless_prepare.py": frozenset({"pipeline"}), "execution/headless/_headless_result.py": frozenset({"recipe"}), + "execution/headless/_headless_terminal.py": frozenset({"pipeline"}), "execution/headless/_managed/_executor.py": frozenset({"pipeline", "recipe"}), "execution/evidence/linux_tracing.py": frozenset({"config"}), "execution/process/__init__.py": frozenset({"config"}), diff --git a/tests/config/test_rollout_config.py b/tests/config/test_rollout_config.py new file mode 100644 index 0000000000..ca20456f82 --- /dev/null +++ b/tests/config/test_rollout_config.py @@ -0,0 +1,158 @@ +"""Validation for the optional rollout declaration in configuration.""" + +from __future__ import annotations + +import copy + +import pytest + +from autoskillit.config import ConfigSchemaError, load_config +from autoskillit.core import dump_yaml_str + +pytestmark = [pytest.mark.layer("config"), pytest.mark.small] + +_VALID_ACTIVE: dict[str, object] = { + "change_id": "prompt-cost-study", + "effective_at": "2026-11-01T18:00:00Z", + "scope": { + "skills": ["make-plan"], + "backends": ["claude-code", "codex"], + "roles": ["planner"], + }, + "expected_effect": "Reduce planning input tokens while preserving requirement coverage.", +} + + +_MISSING = object() + + +def _write_config(tmp_path, config: dict) -> None: + config_dir = tmp_path / ".autoskillit" + config_dir.mkdir() + (config_dir / "config.yaml").write_text(dump_yaml_str(config)) + + +@pytest.mark.parametrize( + ("effective_at", "expected"), + [ + ("2099-11-01T18:00:00Z", "2099-11-01T18:00:00+00:00"), + ("2000-01-01T00:00:00+00:00", "2000-01-01T00:00:00+00:00"), + ], + ids=["future-cutover", "past-cutover"], +) +def test_loads_complete_rollout_active_and_normalizes_utc(tmp_path, effective_at, expected): + active = copy.deepcopy(_VALID_ACTIVE) + active["effective_at"] = effective_at + active["expected_effect"] = " Preserve this prediction text. " + _write_config(tmp_path, {"rollout": {"active": active}}) + + marker = load_config(tmp_path).rollout.active + assert marker is not None + + assert marker == { + **active, + "effective_at": expected, + } + assert marker["expected_effect"] == " Preserve this prediction text. " + assert marker["scope"] == _VALID_ACTIVE["scope"] + + +def test_rollout_validator_copies_raw_scope_lists() -> None: + from autoskillit.config._coercion import _build_rollout_active + + raw_active = copy.deepcopy(_VALID_ACTIVE) + section = {"active": raw_active} + + marker = _build_rollout_active(section, {}) + assert marker is not None + raw_scope = raw_active["scope"] + assert isinstance(raw_scope, dict) + raw_skills = raw_scope["skills"] + raw_backends = raw_scope["backends"] + assert isinstance(raw_skills, list) + assert isinstance(raw_backends, list) + raw_skills.append("later-skill") + raw_backends.clear() + + assert marker["scope"] == { + "skills": ["make-plan"], + "backends": ["claude-code", "codex"], + "roles": ["planner"], + } + + +@pytest.mark.parametrize( + "config", + [ + {}, + {"rollout": {}}, + {"rollout": {"active": None}}, + ], + ids=["missing-section", "missing-active", "null-active"], +) +def test_missing_or_null_rollout_active_is_inactive(tmp_path, config): + _write_config(tmp_path, config) + + assert load_config(tmp_path).rollout.active is None + + +def _invalid_active_cases() -> list[tuple[str, object]]: + cases: list[tuple[str, object]] = [] + + def add(name: str, **changes: object) -> None: + active: dict[str, object] = copy.deepcopy(_VALID_ACTIVE) + for key, value in changes.items(): + if value is _MISSING: + active.pop(key, None) + else: + active[key] = value + cases.append((name, active)) + + cases.append(("empty-active", {})) + cases.append(("partial-active", {"change_id": "only-an-id"})) + add("missing-change-id", change_id=_MISSING) + add("blank-change-id", change_id=" \t") + add("non-string-change-id", change_id=7) + add("missing-expected-effect", expected_effect=_MISSING) + add("null-expected-effect", expected_effect=None) + add("blank-expected-effect", expected_effect=" \t") + add("non-string-expected-effect", expected_effect=7) + add("missing-effective-at", effective_at=_MISSING) + add("malformed-effective-at", effective_at="not-a-date") + add("naive-effective-at", effective_at="2026-11-01T18:00:00") + add("non-utc-effective-at", effective_at="2026-11-01T18:00:00-07:00") + add("non-string-effective-at", effective_at=1700000000) + add("missing-scope", scope=_MISSING) + add("null-scope", scope=None) + add("non-mapping-scope", scope=["make-plan"]) + add("missing-scope-axis", scope={"skills": [], "backends": []}) + add( + "unknown-scope-axis", + scope={"skills": [], "backends": [], "roles": [], "repositories": ["repo"]}, + ) + add("non-list-scope-value", scope={"skills": "make-plan", "backends": [], "roles": []}) + add("non-string-scope-item", scope={"skills": [None], "backends": [], "roles": []}) + add("blank-scope-item", scope={"skills": [" "], "backends": [], "roles": []}) + add("unknown-marker-key", cohort="not-supported") + cases.extend( + [ + ("non-mapping-active-list", []), + ("non-mapping-active-string", "enabled"), + ] + ) + return cases + + +_INVALID_ACTIVE_CASES = _invalid_active_cases() + + +@pytest.mark.parametrize( + ("case", "active"), + _INVALID_ACTIVE_CASES, + ids=[case for case, _active in _INVALID_ACTIVE_CASES], +) +def test_invalid_rollout_active_declarations_raise_schema_error(tmp_path, case, active): + _write_config(tmp_path, {"rollout": {"active": active}}) + + with pytest.raises(ConfigSchemaError): + load_config(tmp_path) diff --git a/tests/contracts/config_key_ledger.txt b/tests/contracts/config_key_ledger.txt index 1c4515e7be..9765be4594 100644 --- a/tests/contracts/config_key_ledger.txt +++ b/tests/contracts/config_key_ledger.txt @@ -110,6 +110,7 @@ report_bug.timeout reset_workspace.command reset_workspace.preserve_dirs review.local_review_rounds +rollout.active run_skill.completion_child_deferral_ceiling_seconds run_skill.completion_drain_timeout run_skill.completion_marker diff --git a/tests/contracts/test_protocol_satisfaction.py b/tests/contracts/test_protocol_satisfaction.py index 3eeecb0778..03b62916f3 100644 --- a/tests/contracts/test_protocol_satisfaction.py +++ b/tests/contracts/test_protocol_satisfaction.py @@ -535,6 +535,8 @@ def test_req_api_001_public_params_present(self): "lifecycle_observation_enabled", "ceiling_seconds", "systemd_scope_enabled", + "rollout", + "logical_run_start_ts", } assert expected == public_params, ( f"run_managed_async public params changed.\n" @@ -560,6 +562,8 @@ def test_req_api_001_public_param_defaults(self): assert p["completion_marker"].default == "" assert p["stale_threshold"].default == 1200 assert p["completion_drain_timeout"].default == 5.0 + assert p["rollout"].default is None + assert p["logical_run_start_ts"].default == "" # ------------------------------------------------------------------ # REQ-API-002: DefaultSubprocessRunner satisfies SubprocessRunner protocol @@ -618,6 +622,8 @@ def test_req_api_002_call_params_match_protocol(self): "lifecycle_observation_enabled", "ceiling_seconds", "systemd_scope_enabled", + "rollout", + "logical_run_start_ts", } assert expected == actual, ( f"DefaultSubprocessRunner.__call__ params changed.\n" diff --git a/tests/core/test_session_index_schema.py b/tests/core/test_session_index_schema.py index 00678c66d0..88b5e6042e 100644 --- a/tests/core/test_session_index_schema.py +++ b/tests/core/test_session_index_schema.py @@ -27,6 +27,7 @@ "backend", "skill_command", "session_type", + "rollout", "subagent_model_outcomes", "evidence_worktree_path", "evidence_worktree_source", @@ -71,6 +72,14 @@ def test_has_nullable_session_type(self): hints = get_type_hints(SessionIndexEntry) assert hints["session_type"] == str | None + def test_has_nullable_rollout_marker(self): + from typing import get_type_hints + + from autoskillit.core.types.results._type_results import RolloutMarker, SessionIndexEntry + + hints = get_type_hints(SessionIndexEntry) + assert hints["rollout"] == RolloutMarker | None + def test_canonical_cache_fields(self): """SessionIndexEntry must use canonical cache field names, not v1 API names.""" from autoskillit.core.types.results._type_results import SessionIndexEntry diff --git a/tests/core/test_session_index_schema_version_lock.py b/tests/core/test_session_index_schema_version_lock.py index edd97e7001..b530feadd5 100644 --- a/tests/core/test_session_index_schema_version_lock.py +++ b/tests/core/test_session_index_schema_version_lock.py @@ -12,7 +12,7 @@ pytestmark = [pytest.mark.layer("core"), pytest.mark.small] -_EXPECTED = (16, "9584465d5e3ca791680c05a0698482e23561ef03533c209a40380d8506580eb3") +_EXPECTED = (17, "02755757afa42542ffde3c6ac1be6d3e7a2116d57069447932eac4380dd1fdb1") def test_session_index_schema_version_matches_field_digest() -> None: diff --git a/tests/core/test_type_results_records.py b/tests/core/test_type_results_records.py index c232fe5a97..c105b96afe 100644 --- a/tests/core/test_type_results_records.py +++ b/tests/core/test_type_results_records.py @@ -39,6 +39,7 @@ "EvidenceWorktreeSource", "ValidatedAddDir", "ValidatedWorktreePath", + "RolloutMarker", ) INTERNAL_INDEX_NAMES = ( "TokenUsageFileEntry", diff --git a/tests/execution/conftest.py b/tests/execution/conftest.py index d286cc74e4..83315773d6 100644 --- a/tests/execution/conftest.py +++ b/tests/execution/conftest.py @@ -40,6 +40,11 @@ from tests._helpers import make_tracing_config from tests.fixtures.codex import FLAT_ERROR_MODEL_CAPACITY, TURN_FAILED_MODEL_CAPACITY + +def _unmarked_rollout_snapshot() -> tuple[None, str]: + return None, "2000-01-01T00:00:00+00:00" + + # Strip regex metacharacters (e.g. \b word boundaries) to yield plain test signal strings. CODEX_API_ERROR_SIGNAL_STRINGS: tuple[str, ...] = tuple( p.pattern.replace(r"\b", "") for p in _CODEX_API_ERROR_PATTERNS diff --git a/tests/execution/headless/test_codex_attempt_retention.py b/tests/execution/headless/test_codex_attempt_retention.py index 204b8b851a..31e3ac8064 100644 --- a/tests/execution/headless/test_codex_attempt_retention.py +++ b/tests/execution/headless/test_codex_attempt_retention.py @@ -27,7 +27,7 @@ from autoskillit.execution.backends import codex as codex_module from autoskillit.execution.backends._codex_session_storage import CodexSessionStore from autoskillit.execution.backends.codex import CodexBackend -from tests.execution.conftest import _launch_inputs, _mock_backend +from tests.execution.conftest import _launch_inputs, _mock_backend, _unmarked_rollout_snapshot from tests.fakes import FakePluginArtifactAuthority pytestmark = [pytest.mark.layer("execution"), pytest.mark.medium] @@ -189,6 +189,7 @@ async def _run_generated_attempt( completion_marker="%%DONE%%", stale_threshold=1.0, completion_drain_timeout=1.0, + capture_rollout_on_launch=_unmarked_rollout_snapshot, natural_exit_grace_seconds=1.0, linux_tracing_config=None, idle_output_timeout=None, @@ -303,6 +304,7 @@ def build_resume_cmd(**kwargs: object) -> CmdSpec: session_home=str(first_home), launch_resolver=nudge_resolver, launch_preparation=nudge_preparation, + capture_rollout_on_launch=_unmarked_rollout_snapshot, natural_exit_grace_seconds=1.0, attempt=2, ceiling_seconds=_EXECUTION_CEILING_SECONDS, @@ -415,6 +417,7 @@ def build_with_retained_binding(binding, _extras, _attempt_id=None) -> CmdSpec: completion_marker="%%DONE%%", stale_threshold=1.0, completion_drain_timeout=1.0, + capture_rollout_on_launch=_unmarked_rollout_snapshot, natural_exit_grace_seconds=1.0, linux_tracing_config=None, idle_output_timeout=None, @@ -488,6 +491,7 @@ def build_resume_cmd(**kwargs: object) -> CmdSpec: session_home=str(home), launch_resolver=nudge_resolver, launch_preparation=nudge_preparation, + capture_rollout_on_launch=_unmarked_rollout_snapshot, natural_exit_grace_seconds=1.0, attempt=2, ceiling_seconds=_EXECUTION_CEILING_SECONDS, diff --git a/tests/execution/headless/test_headless_recovery.py b/tests/execution/headless/test_headless_recovery.py index fd2e28f495..b395e7107a 100644 --- a/tests/execution/headless/test_headless_recovery.py +++ b/tests/execution/headless/test_headless_recovery.py @@ -27,7 +27,7 @@ from autoskillit.core.types.results._type_results import WriteEvidence from autoskillit.execution.headless._managed import _ManagedLineageObserver from autoskillit.execution.session.turn_usage import build_turn_token_entry -from tests.execution.conftest import _launch_inputs, _mock_backend +from tests.execution.conftest import _launch_inputs, _mock_backend, _unmarked_rollout_snapshot from tests.fakes import FakeManagedHeadlessSessionLineageStore pytestmark = [pytest.mark.layer("execution"), pytest.mark.small] @@ -138,6 +138,7 @@ async def _run_turn_usage_nudge( retry_reason=RetryReason.EARLY_STOP, launch_resolver=launch_resolver, launch_preparation=launch_preparation, + capture_rollout_on_launch=_unmarked_rollout_snapshot, natural_exit_grace_seconds=3.0, ) assert result is not None @@ -175,6 +176,7 @@ class TestNudgePtyMode: @pytest.mark.anyio async def test_nudge_passes_pty_mode_for_claude_backend(self, tmp_path: Path) -> None: """_attempt_contract_nudge passes pty_mode=True to runner for ClaudeCode backend.""" + from autoskillit.config import LinuxTracingConfig from autoskillit.execution.headless._headless_launch import _attempt_contract_nudge from tests.execution.conftest import _mock_backend from tests.fakes import MockSubprocessRunner @@ -226,6 +228,15 @@ async def test_nudge_passes_pty_mode_for_claude_backend(self, tmp_path: Path) -> termination=TerminationReason.NATURAL_EXIT, pid=0, ) + rollout_marker = { + "change_id": "rollout-nudge", + "effective_at": "2026-10-10T12:00:00+00:00", + "scope": {"skills": ["audit"], "backends": ["claude-code"], "roles": []}, + "expected_effect": "Nudge retains the admitted snapshot", + } + logical_start = "2026-10-10T12:00:01+00:00" + capture_rollout = Mock(return_value=(rollout_marker, logical_start)) + linux_tracing_config = LinuxTracingConfig(tmpfs_path=str(tmp_path)) await _attempt_contract_nudge( skill_result=skill_result, @@ -239,6 +250,8 @@ async def test_nudge_passes_pty_mode_for_claude_backend(self, tmp_path: Path) -> retry_reason=RetryReason.EARLY_STOP, launch_resolver=launch_resolver, launch_preparation=launch_preparation, + capture_rollout_on_launch=capture_rollout, + linux_tracing_config=linux_tracing_config, natural_exit_grace_seconds=3.0, ) @@ -247,6 +260,11 @@ async def test_nudge_passes_pty_mode_for_claude_backend(self, tmp_path: Path) -> f"Expected last_pty_mode=True for ClaudeCode backend (pty_required=True), " f"got {mock_runner.last_pty_mode!r}" ) + nudge_kwargs = mock_runner.call_args_list[0][3] + capture_rollout.assert_called_once_with() + assert nudge_kwargs["rollout"] == rollout_marker + assert nudge_kwargs["logical_run_start_ts"] == logical_start + assert nudge_kwargs["linux_tracing_config"] is linux_tracing_config @pytest.mark.anyio async def test_nudge_respects_pty_override_false(self, tmp_path: Path) -> None: @@ -315,6 +333,7 @@ async def test_nudge_respects_pty_override_false(self, tmp_path: Path) -> None: pty_override=False, launch_resolver=launch_resolver, launch_preparation=launch_preparation, + capture_rollout_on_launch=_unmarked_rollout_snapshot, natural_exit_grace_seconds=3.0, ) @@ -429,6 +448,7 @@ async def __call__(self, *args, **kwargs): managed_lineage_observer=lineage_observer, launch_resolver=launch_resolver, launch_preparation=launch_preparation, + capture_rollout_on_launch=_unmarked_rollout_snapshot, natural_exit_grace_seconds=3.0, ) @@ -514,6 +534,7 @@ async def test_nudge_passes_backend_line_driver_to_runner(self, tmp_path: Path) retry_reason=RetryReason.EARLY_STOP, launch_resolver=launch_resolver, launch_preparation=launch_preparation, + capture_rollout_on_launch=_unmarked_rollout_snapshot, natural_exit_grace_seconds=3.0, ) @@ -603,6 +624,7 @@ def acquire_launch_binding(self, *, backend, load_mode): retained_binding=retained_binding, launch_resolver=launch_resolver, launch_preparation=launch_preparation, + capture_rollout_on_launch=_unmarked_rollout_snapshot, natural_exit_grace_seconds=3.0, ) @@ -680,6 +702,7 @@ async def test_nudge_passes_managed_skill_catalog_to_build_resume_cmd( managed_skill_catalog=catalog, launch_resolver=launch_resolver, launch_preparation=launch_preparation, + capture_rollout_on_launch=_unmarked_rollout_snapshot, natural_exit_grace_seconds=3.0, ) @@ -809,6 +832,7 @@ async def test_nudge_skips_for_block_delimiter_patterns(tmp_path: Path) -> None: result_parser=result_parser, launch_resolver=launch_resolver, launch_preparation=launch_preparation, + capture_rollout_on_launch=_unmarked_rollout_snapshot, natural_exit_grace_seconds=3.0, ) diff --git a/tests/execution/test_headless_capture_diagnostics.py b/tests/execution/test_headless_capture_diagnostics.py index 73d0101e3a..9b8c43fc34 100644 --- a/tests/execution/test_headless_capture_diagnostics.py +++ b/tests/execution/test_headless_capture_diagnostics.py @@ -99,6 +99,7 @@ def __init__(self, *, runner: Any, log_dir: str) -> None: completion_child_deferral_ceiling_seconds=1.0, ), providers=SimpleNamespace(provider_retry_limit=0), + rollout=SimpleNamespace(active=None), linux_tracing=SimpleNamespace( log_dir=log_dir, max_sessions=20, diff --git a/tests/execution/test_headless_dispatch.py b/tests/execution/test_headless_dispatch.py index 3e8562382a..f349369435 100644 --- a/tests/execution/test_headless_dispatch.py +++ b/tests/execution/test_headless_dispatch.py @@ -183,10 +183,11 @@ def close(self) -> None: monkeypatch.setattr(headless_launch, "logger", logger) with pytest.raises(RuntimeError) as caught: - with headless_launch._plugin_launch_binding( + with headless_launch.plugin_launch_binding_scope( authority=authority, backend=ClaudeCodeBackend(), load_mode=PluginLoadMode.EXPLICIT_PLUGIN_DIR, + on_suppressed_close_error=headless_launch._report_plugin_binding_close_failure, ): raise primary diff --git a/tests/execution/test_headless_enum_recovery.py b/tests/execution/test_headless_enum_recovery.py index b9944f5963..57dfcf5efc 100644 --- a/tests/execution/test_headless_enum_recovery.py +++ b/tests/execution/test_headless_enum_recovery.py @@ -28,6 +28,10 @@ pytestmark = [pytest.mark.layer("execution"), pytest.mark.small] +def _unmarked_rollout_snapshot() -> tuple[None, str]: + return None, "2000-01-01T00:00:00+00:00" + + def _make_plan_contract( *, write_behavior: str | None = "conditional", @@ -366,6 +370,7 @@ async def test_enum_nudge_success_requires_pattern_match(self, tmp_path): skill_contract=contract, launch_resolver=launch_resolver, launch_preparation=launch_preparation, + capture_rollout_on_launch=_unmarked_rollout_snapshot, natural_exit_grace_seconds=3.0, ) assert patched is not None @@ -388,6 +393,7 @@ async def test_enum_nudge_success_requires_pattern_match(self, tmp_path): skill_contract=contract, launch_resolver=launch_resolver, launch_preparation=launch_preparation, + capture_rollout_on_launch=_unmarked_rollout_snapshot, natural_exit_grace_seconds=3.0, ) assert rejected is None diff --git a/tests/execution/test_headless_execute.py b/tests/execution/test_headless_execute.py index b93cbe5e75..211554fa19 100644 --- a/tests/execution/test_headless_execute.py +++ b/tests/execution/test_headless_execute.py @@ -4,6 +4,7 @@ import json import os +from datetime import UTC, datetime, tzinfo from pathlib import Path from unittest.mock import patch @@ -11,7 +12,7 @@ import structlog import autoskillit.execution.headless._headless_execute as _patch_headless__headless_execute -from autoskillit.core import CmdSpec, ValidatedAddDir +from autoskillit.core import CmdSpec, RolloutMarker, ValidatedAddDir from autoskillit.core.types import SubprocessResult, TerminationReason from tests.execution.conftest import _sink_env @@ -38,6 +39,369 @@ def _success_result() -> SubprocessResult: ) +class _ControlledDateTime(datetime): + current: datetime = datetime(2026, 10, 10, 12, 0, 0, 499999, tzinfo=UTC) + + @classmethod + def now(cls, tz: tzinfo | None = None) -> datetime: + value = cls.current.replace(tzinfo=None) if tz is None else cls.current.astimezone(tz) + return cls.fromisoformat(value.isoformat()) + + +def _rollout_marker(change_id: str) -> RolloutMarker: + return { + "change_id": change_id, + "effective_at": "2026-10-10T12:00:00.500000+00:00", + "scope": {"skills": ["audit"], "backends": ["claude-code", "codex"], "roles": []}, + "expected_effect": f"Expected effect for {change_id}", + } + + +async def _run_rollout_execution( + ctx, + tmp_path: Path, + *, + backend_name: str = "claude-code", + pre_spawn_admission=None, +): + from autoskillit.execution.headless import _execute_claude_headless + from tests.execution.conftest import _launch_preparation, _mock_backend + + backend = _mock_backend(pty_required=True, channel_b_capable=True) + backend.name = backend_name + ctx.backend = backend + return await _execute_claude_headless( + lambda _binding, extras, _attempt_id=None: CmdSpec( + cmd=(backend_name, "--test"), env=dict(extras or {}) + ), + str(tmp_path), + ctx, + timeout=30.0, + stale_threshold=5.0, + step_name="rollout-test", + launch_resolver=ctx.launch_resolver, + launch_preparation=_launch_preparation(ctx, cwd=str(tmp_path)), + pre_spawn_admission=pre_spawn_admission, + ) + + +def _capture_rollout_flushes(monkeypatch, ctx, tmp_path: Path) -> list[dict[str, object]]: + import autoskillit.execution as execution + import autoskillit.execution.headless._headless_execute as execute_module + import autoskillit.execution.session_log.session_log as session_log + + log_dir = str(tmp_path / "rollout-logs") + ctx.config.linux_tracing.log_dir = log_dir + _install_fake_sink( + monkeypatch, + execute_module, + [], + log_dir, + close_raises=False, + ) + flushed: list[dict[str, object]] = [] + original_flush = session_log.flush_session_log + + def capture_flush(**kwargs): + flushed.append(kwargs) + original_flush(**kwargs) + + monkeypatch.setattr(session_log, "flush_session_log", capture_flush) + monkeypatch.setattr(execution, "flush_session_log", capture_flush) + return flushed + + +def _assert_rollout_artifacts(ctx, rollout: RolloutMarker | None, start_ts: str) -> None: + log_root = Path(ctx.config.linux_tracing.log_dir) + summary_paths = list(log_root.rglob("summary.json")) + assert len(summary_paths) == 1 + summary = json.loads(summary_paths[0].read_text()) + index_rows = [ + json.loads(line) + for line in (log_root / "sessions.jsonl").read_text().splitlines() + if line.strip() + ] + assert len(index_rows) == 1 + assert summary["rollout"] == rollout + assert summary["start_ts"] == start_ts + assert index_rows[0]["rollout"] == rollout + assert index_rows[0]["timestamp"] == start_ts + + +@pytest.mark.anyio +@pytest.mark.parametrize("backend_name", ["claude-code", "codex"]) +async def test_rollout_snapshot_is_taken_after_admission_for_both_backends( + minimal_ctx, tmp_path: Path, monkeypatch, backend_name: str +) -> None: + import autoskillit.execution.headless._headless_execute as execute_module + from tests.fakes import MockSubprocessRunner + + before_cutover = datetime.fromisoformat("2026-10-10T12:00:00.499999+00:00") + after_cutover = datetime.fromisoformat("2026-10-10T12:00:00.500001+00:00") + marker_a = _rollout_marker("rollout-a") + marker_b = _rollout_marker("rollout-b") + minimal_ctx.config.rollout.active = marker_a + monkeypatch.setattr(_ControlledDateTime, "current", before_cutover) + monkeypatch.setattr(execute_module, "datetime", _ControlledDateTime) + + runner = MockSubprocessRunner() + runner.set_default(_success_result()) + minimal_ctx.runner = runner + flushed = _capture_rollout_flushes(monkeypatch, minimal_ctx, tmp_path) + admission_calls = 0 + + async def admit_after_cutover(_contract): + nonlocal admission_calls + admission_calls += 1 + minimal_ctx.config.rollout.active = marker_b + monkeypatch.setattr(_ControlledDateTime, "current", after_cutover) + return None + + await _run_rollout_execution( + minimal_ctx, + tmp_path, + backend_name=backend_name, + pre_spawn_admission=admit_after_cutover, + ) + + assert admission_calls == 1 + assert len(runner.call_args_list) == 1 + runner_kwargs = runner.call_args_list[0][3] + assert runner_kwargs["rollout"] == marker_b + assert runner_kwargs["logical_run_start_ts"] == after_cutover.isoformat() + assert datetime.fromisoformat(runner_kwargs["logical_run_start_ts"]) >= datetime.fromisoformat( + marker_b["effective_at"] + ) + assert flushed[0]["rollout"] == marker_b + assert flushed[0]["start_ts"] == after_cutover.isoformat() + _assert_rollout_artifacts(minimal_ctx, marker_b, after_cutover.isoformat()) + + +@pytest.mark.anyio +async def test_admitted_inactive_run_uses_logical_start_over_later_result_start( + minimal_ctx, tmp_path: Path, monkeypatch +) -> None: + import autoskillit.execution.headless._headless_execute as execute_module + import autoskillit.execution.headless._headless_terminal as terminal + from tests.fakes import MockSubprocessRunner + + operation_start = datetime.fromisoformat("2026-10-10T12:00:00.499999+00:00") + logical_start = datetime.fromisoformat("2026-10-10T12:00:00.500001+00:00") + later_result_start = datetime.fromisoformat("2026-10-10T12:00:00.500003+00:00") + minimal_ctx.config.rollout.active = None + monkeypatch.setattr(_ControlledDateTime, "current", operation_start) + monkeypatch.setattr(execute_module, "datetime", _ControlledDateTime) + + runner = MockSubprocessRunner() + runner.set_default(_success_result()) + minimal_ctx.runner = runner + flushed = _capture_rollout_flushes(monkeypatch, minimal_ctx, tmp_path) + original_builder = terminal.build_terminal_flush_kwargs + observed_result_starts: list[str] = [] + + def set_later_result_start(**kwargs): + result = kwargs["result"] + assert result is not None + observed_result_starts.append(result.start_ts) + result.start_ts = later_result_start.isoformat() + return original_builder(**kwargs) + + monkeypatch.setattr(terminal, "build_terminal_flush_kwargs", set_later_result_start) + + async def admit(_contract): + monkeypatch.setattr(_ControlledDateTime, "current", logical_start) + return None + + await _run_rollout_execution(minimal_ctx, tmp_path, pre_spawn_admission=admit) + + runner_kwargs = runner.call_args_list[0][3] + assert runner_kwargs["rollout"] is None + assert runner_kwargs["logical_run_start_ts"] == logical_start.isoformat() + assert observed_result_starts == [operation_start.isoformat()] + assert later_result_start > logical_start + assert flushed[0]["rollout"] is None + assert flushed[0]["start_ts"] == logical_start.isoformat() + + +@pytest.mark.anyio +async def test_rollout_snapshot_survives_midrun_config_mutation_and_independent_run( + minimal_ctx, tmp_path: Path, monkeypatch +) -> None: + from copy import deepcopy + + import autoskillit.execution.headless._headless_execute as execute_module + from tests.fakes import MockSubprocessRunner + + marker_a = _rollout_marker("rollout-a") + marker_b = _rollout_marker("rollout-b") + minimal_ctx.config.rollout.active = deepcopy(marker_a) + monkeypatch.setattr( + _ControlledDateTime, + "current", + datetime.fromisoformat("2026-10-10T12:00:01+00:00"), + ) + monkeypatch.setattr(execute_module, "datetime", _ControlledDateTime) + + class MutatingRunner(MockSubprocessRunner): + rollout_calls: list[dict[str, object]] + + def __init__(self) -> None: + super().__init__() + self.rollout_calls = [] + self.set_default(_success_result()) + + async def __call__(self, *args, **kwargs): + self.rollout_calls.append(kwargs) + if len(self.rollout_calls) == 1: + configured = minimal_ctx.config.rollout.active + assert configured is not None + configured["scope"]["skills"].clear() + minimal_ctx.config.rollout.active = None + monkeypatch.setattr( + _ControlledDateTime, + "current", + datetime.fromisoformat("2026-10-10T12:00:02+00:00"), + ) + return await super().__call__(*args, **kwargs) + + runner = MutatingRunner() + minimal_ctx.runner = runner + flushed = _capture_rollout_flushes(monkeypatch, minimal_ctx, tmp_path) + + await _run_rollout_execution(minimal_ctx, tmp_path) + + assert runner.rollout_calls[0]["rollout"] == marker_a + assert runner.rollout_calls[0]["logical_run_start_ts"] == "2026-10-10T12:00:01+00:00" + assert flushed[0]["rollout"] == marker_a + assert flushed[0]["start_ts"] == runner.rollout_calls[0]["logical_run_start_ts"] + + minimal_ctx.config.rollout.active = marker_b + await _run_rollout_execution(minimal_ctx, tmp_path) + + assert runner.rollout_calls[1]["rollout"] == marker_b + assert runner.rollout_calls[1]["logical_run_start_ts"] == "2026-10-10T12:00:02+00:00" + assert flushed[1]["rollout"] == marker_b + assert flushed[1]["start_ts"] == runner.rollout_calls[1]["logical_run_start_ts"] + + +@pytest.mark.anyio +async def test_contract_nudge_reuses_executor_rollout_snapshot_after_config_changes( + minimal_ctx, tmp_path: Path, monkeypatch +) -> None: + import autoskillit.execution.headless._headless_execute as execute_module + from autoskillit.core import RetryReason, SkillResult + from tests.fakes import MockSubprocessRunner + + marker_a = _rollout_marker("rollout-a") + marker_b = _rollout_marker("rollout-b") + minimal_ctx.config.rollout.active = marker_a + monkeypatch.setattr( + _ControlledDateTime, + "current", + datetime.fromisoformat("2026-10-10T12:00:03+00:00"), + ) + monkeypatch.setattr(execute_module, "datetime", _ControlledDateTime) + + class MutatingRunner(MockSubprocessRunner): + observed_kwargs: dict[str, object] | None = None + + async def __call__(self, *args, **kwargs): + self.observed_kwargs = kwargs + minimal_ctx.config.rollout.active = marker_b + monkeypatch.setattr( + _ControlledDateTime, + "current", + datetime.fromisoformat("2026-10-10T12:00:04+00:00"), + ) + return await super().__call__(*args, **kwargs) + + runner = MutatingRunner() + runner.set_default(_success_result()) + minimal_ctx.runner = runner + flushed = _capture_rollout_flushes(monkeypatch, minimal_ctx, tmp_path) + retry_result = SkillResult( + success=False, + result="missing completion marker", + session_id="nudge-session", + subtype="contract_recovery", + is_error=False, + exit_code=1, + needs_retry=True, + retry_reason=RetryReason.CONTRACT_RECOVERY, + stderr="", + ) + nudge_snapshots: list[tuple[RolloutMarker | None, str]] = [] + nudge_configs: list[object] = [] + + monkeypatch.setattr( + execute_module, + "_build_skill_result", + lambda *_args, **_kwargs: retry_result, + ) + + async def nudge_with_shared_snapshot(*_args, **kwargs): + nudge_snapshots.append(kwargs["capture_rollout_on_launch"]()) + nudge_configs.append(kwargs["linux_tracing_config"]) + return SkillResult( + success=True, + result="recovered", + session_id="nudge-session", + subtype="success", + is_error=False, + exit_code=0, + needs_retry=False, + retry_reason=RetryReason.NONE, + stderr="", + ) + + monkeypatch.setattr(execute_module, "_attempt_contract_nudge", nudge_with_shared_snapshot) + + await _run_rollout_execution(minimal_ctx, tmp_path) + + assert runner.observed_kwargs is not None + main_snapshot = ( + runner.observed_kwargs["rollout"], + runner.observed_kwargs["logical_run_start_ts"], + ) + assert main_snapshot == (marker_a, "2026-10-10T12:00:03+00:00") + assert nudge_snapshots == [main_snapshot] + assert nudge_configs == [minimal_ctx.config.linux_tracing] + assert flushed[0]["rollout"] == marker_a + assert flushed[0]["start_ts"] == main_snapshot[1] + + +@pytest.mark.anyio +async def test_launched_runner_failure_keeps_rollout_marker_in_flush( + minimal_ctx, tmp_path: Path, monkeypatch +) -> None: + from tests.fakes import MockSubprocessRunner + + marker = _rollout_marker("rollout-failure") + minimal_ctx.config.rollout.active = marker + + class FailingRunner(MockSubprocessRunner): + observed_kwargs: dict[str, object] | None = None + + async def __call__(self, *args, **kwargs): + self.observed_kwargs = kwargs + raise RuntimeError("runner failed after launch") + + runner = FailingRunner() + minimal_ctx.runner = runner + flushed = _capture_rollout_flushes(monkeypatch, minimal_ctx, tmp_path) + + result = await _run_rollout_execution(minimal_ctx, tmp_path) + + assert result.subtype == "crashed" + assert runner.observed_kwargs is not None + assert runner.observed_kwargs["rollout"] == marker + logical_start = runner.observed_kwargs["logical_run_start_ts"] + assert isinstance(logical_start, str) and logical_start + assert flushed[0]["rollout"] == marker + assert flushed[0]["start_ts"] == logical_start + _assert_rollout_artifacts(minimal_ctx, marker, logical_start) + + def _install_fake_sink( monkeypatch, execute_module, diff --git a/tests/execution/test_linux_tracing.py b/tests/execution/test_linux_tracing.py index 032851e2ee..d480126f82 100644 --- a/tests/execution/test_linux_tracing.py +++ b/tests/execution/test_linux_tracing.py @@ -6,10 +6,12 @@ import os import sys from datetime import datetime +from unittest.mock import Mock import anyio import pytest +from tests._helpers import make_rollout_marker from tests.conftest import production_interpreter_env pytestmark = [ @@ -450,21 +452,36 @@ async def test_start_linux_tracing_writes_enrollment_sidecar(tmp_path): import anyio from autoskillit.execution.evidence.linux_tracing import ( + read_enrollment, start_linux_tracing, trace_target_from_pid, ) from tests._helpers import make_tracing_config cfg = make_tracing_config(enabled=True, proc_interval=0.1, tmpfs_path=str(tmp_path)) + rollout = make_rollout_marker() + logical_run_start_ts = "2026-10-10T19:00:00+00:00" async with anyio.create_task_group() as tg: proc = await anyio.open_process(["sleep", "2"]) - handle = start_linux_tracing(trace_target_from_pid(proc.pid), cfg, tg) + handle = start_linux_tracing( + trace_target_from_pid(proc.pid), + cfg, + tg, + rollout=rollout, + logical_run_start_ts=logical_run_start_ts, + ) 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"] == 2 + assert data["schema_version"] == 3 assert data["pid"] == proc.pid + assert data["rollout"] == rollout + assert data["logical_run_start_ts"] == logical_run_start_ts + record = read_enrollment(enrollment) + assert record is not None + assert record.rollout == rollout + assert record.logical_run_start_ts == logical_run_start_ts try: handle.stop() finally: @@ -472,6 +489,48 @@ async def test_start_linux_tracing_writes_enrollment_sidecar(tmp_path): assert not enrollment.exists(), "Enrollment must be deleted by stop()" +@pytest.mark.parametrize("schema_version", [1, 2]) +def test_read_enrollment_defaults_rollout_fields_for_historical_schemas(tmp_path, schema_version): + from autoskillit.execution.evidence.linux_tracing import read_enrollment + + data = {"schema_version": schema_version, "pid": 12345} + if schema_version == 2: + data["comm"] = "claude" + enrollment = tmp_path / "enrollment.json" + enrollment.write_text(json.dumps(data)) + + record = read_enrollment(enrollment) + + assert record is not None + assert record.schema_version == schema_version + assert record.rollout is None + assert record.logical_run_start_ts == "" + + +@pytest.mark.parametrize("contents", [None, "{"]) +def test_read_enrollment_logs_invalid_sidecar_but_not_missing(tmp_path, monkeypatch, contents): + from autoskillit.execution.evidence import linux_tracing + + enrollment = tmp_path / "enrollment.json" + if contents is not None: + enrollment.write_text(contents) + logger = Mock() + monkeypatch.setattr(linux_tracing, "logger", logger) + + assert linux_tracing.read_enrollment(enrollment) is None + + if contents is None: + logger.warning.assert_not_called() + else: + logger.warning.assert_called_once() + message, path, error_type, error = logger.warning.call_args.args + assert "Failed to read enrollment sidecar" in message + assert path == enrollment + assert error_type == "JSONDecodeError" + assert isinstance(error, json.JSONDecodeError) + assert str(error) + + # --- LinuxTracingConfig guard tests --- diff --git a/tests/execution/test_model_backend_launch_contract.py b/tests/execution/test_model_backend_launch_contract.py index a7e46f5125..76d3e4a061 100644 --- a/tests/execution/test_model_backend_launch_contract.py +++ b/tests/execution/test_model_backend_launch_contract.py @@ -9,7 +9,7 @@ from __future__ import annotations from typing import Any -from unittest.mock import AsyncMock, patch +from unittest.mock import AsyncMock, Mock, patch import pytest @@ -166,7 +166,7 @@ async def test_retired_codex_model_fails_during_real_command_preparation_before_ """The real Codex command builder rejects retired targets before runner invocation.""" from autoskillit.execution.headless._headless_launch import _run_headless_attempt - kwargs, runner = _build_real_codex_attempt_kwargs_for_retired_model(model_id, tmp_path) + kwargs, runner = _build_codex_attempt_kwargs(model_id, tmp_path) with pytest.raises(ValueError): await _run_headless_attempt(kwargs.pop("build_spec"), runner=runner, **kwargs) @@ -174,15 +174,38 @@ async def test_retired_codex_model_fails_during_real_command_preparation_before_ runner.assert_not_awaited() -def _build_real_codex_attempt_kwargs_for_retired_model( +@pytest.mark.anyio +async def test_rejected_admission_does_not_capture_rollout_or_invoke_runner(tmp_path) -> None: + from autoskillit.core import CandidatePreSpawnRejection + from autoskillit.execution.headless._headless_launch import _run_headless_attempt + + kwargs, runner = _build_codex_attempt_kwargs("gpt-6-sol", tmp_path) + capture = Mock(return_value=(None, "2000-01-01T00:00:00+00:00")) + kwargs["capture_rollout_on_launch"] = capture + + async def reject(contract): + return CandidatePreSpawnRejection( + reason="candidate is not admitted", + attempted_contract=contract, + ) + + kwargs["pre_spawn_admission"] = reject + build_spec = kwargs.pop("build_spec") + result = await _run_headless_attempt(build_spec, runner=runner, **kwargs) + + assert isinstance(result, CandidatePreSpawnRejection) + capture.assert_not_called() + runner.assert_not_awaited() + + +def _build_codex_attempt_kwargs( model_id: str, tmp_path, ) -> tuple[dict[str, Any], AsyncMock]: - """Build the kwargs dict for one retired-Codex-model launch attempt. + """Build the kwargs dict for one real Codex launch attempt. A new required parameter on ``_run_headless_attempt`` will break this helper - at its single definition site, which is the desired locality for a contract - test that pins the function's full parameter surface. + at this single fixture site. """ from autoskillit.core import PluginLoadMode from autoskillit.execution.backends import CodexBackend @@ -192,7 +215,9 @@ def _build_real_codex_attempt_kwargs_for_retired_model( runner = AsyncMock() def build_spec(_binding, _extras, _managed_attempt_id=None): - return backend.build_headless_cmd("test prompt", model=model_id) + return backend.build_headless_cmd( + "test prompt", model=model_id, generated_home=tmp_path / "codex-home" + ) return ( { @@ -201,6 +226,7 @@ def build_spec(_binding, _extras, _managed_attempt_id=None): "launch_resolver": resolver, "launch_preparation": preparation, "expected_launch_contract": None, + "capture_rollout_on_launch": lambda: (None, "2000-01-01T00:00:00+00:00"), "plugin_authority": None, "plugin_load_mode": PluginLoadMode.NONE, "provider_extras": None, diff --git a/tests/execution/test_process_run.py b/tests/execution/test_process_run.py index f81a966e48..ff80039375 100644 --- a/tests/execution/test_process_run.py +++ b/tests/execution/test_process_run.py @@ -338,6 +338,42 @@ def test_runner_pass_fds_defaults_match_protocol(self): == () ) + @pytest.mark.anyio + async def test_default_runner_forwards_rollout_and_logical_run_start( + self, monkeypatch, tmp_path + ): + from autoskillit.execution.process import DefaultSubprocessRunner + from tests.conftest import _make_result + + rollout = { + "change_id": "prompt-cost-study", + "effective_at": "2026-11-01T18:00:00+00:00", + "scope": {"skills": ["make-plan"], "backends": ["codex"], "roles": ["planner"]}, + "expected_effect": "Reduce planning input tokens while preserving coverage.", + } + logical_run_start_ts = "2026-10-10T19:00:00+00:00" + captured: dict[str, object] = {} + expected = _make_result(returncode=0) + + async def fake_run_managed_async(_cmd, **kwargs): + captured.update(kwargs) + return expected + + import autoskillit.execution.process as process + + monkeypatch.setattr(process, "run_managed_async", fake_run_managed_async) + result = await DefaultSubprocessRunner()( + ["python", "-c", "pass"], + cwd=tmp_path, + timeout=10, + rollout=rollout, + logical_run_start_ts=logical_run_start_ts, + ) + + assert result is expected + assert captured["rollout"] is rollout + assert captured["logical_run_start_ts"] == logical_run_start_ts + class TestTracingStopOnException: """Verify tracing_handle.stop() is called on BaseException in run_managed_async.""" @@ -384,6 +420,35 @@ def patched_stop(self_handle: LinuxTracingHandle) -> list: assert len(stop_called) >= 1 +@pytest.mark.anyio +@pytest.mark.skipif(sys.platform != "linux", reason="Linux-only tracing") +async def test_run_managed_async_forwards_rollout_to_trace_enrollment(monkeypatch, tmp_path): + from autoskillit.execution.evidence import linux_tracing + from tests._helpers import make_rollout_marker, make_tracing_config + + rollout = make_rollout_marker(backends=("codex",)) + logical_run_start_ts = "2026-10-10T19:00:00+00:00" + forwarded: list[tuple[object, str]] = [] + + def capture_enrollment(*, rollout=None, logical_run_start_ts="", **_kwargs): + forwarded.append((rollout, logical_run_start_ts)) + + monkeypatch.setattr(linux_tracing, "start_linux_tracing", capture_enrollment) + cfg = make_tracing_config(enabled=True, proc_interval=0.1, tmpfs_path=str(tmp_path)) + + result = await run_managed_async( + [sys.executable, "-c", "pass"], + cwd=tmp_path, + timeout=10, + linux_tracing_config=cfg, + rollout=rollout, + logical_run_start_ts=logical_run_start_ts, + ) + + assert result.returncode == 0 + assert forwarded == [(rollout, logical_run_start_ts)] + + class TestManagedProcessOwnership: """The subprocess boundary, not a server supervisor, owns process cleanup.""" diff --git a/tests/execution/test_recording.py b/tests/execution/test_recording.py index b9a00887ad..dcd6f25130 100644 --- a/tests/execution/test_recording.py +++ b/tests/execution/test_recording.py @@ -27,6 +27,7 @@ _detect_backend_format, _extract_model, ) +from tests._helpers import make_rollout_marker from tests.conftest import _make_result from tests.execution.backends._plugin_binding import plugin_binding from tests.fakes import MockSubprocessRunner @@ -1036,6 +1037,8 @@ async def test_recording_runner_forwards_process_lifecycle_callbacks(tmp_path: P ) on_spawned = Mock() on_reaped = Mock() + rollout = make_rollout_marker(backends=("codex",)) + logical_run_start_ts = "2026-10-10T19:00:00+00:00" await runner( ["codex", "exec", "do something"], @@ -1044,11 +1047,15 @@ async def test_recording_runner_forwards_process_lifecycle_callbacks(tmp_path: P env={"SCENARIO_STEP_NAME": "recorded"}, on_process_spawned=on_spawned, on_process_reaped=on_reaped, + rollout=rollout, + logical_run_start_ts=logical_run_start_ts, ) forwarded = inner.call_args_list[0][3] assert forwarded["on_process_spawned"] is on_spawned assert forwarded["on_process_reaped"] is on_reaped + assert forwarded["rollout"] is rollout + assert forwarded["logical_run_start_ts"] == logical_run_start_ts @pytest.mark.anyio @@ -1098,6 +1105,7 @@ async def test_replay_runner_never_invokes_process_lifecycle_callbacks(tmp_path: ) on_spawned = Mock() on_reaped = Mock() + rollout = make_rollout_marker(backends=("codex",)) await runner( ["codex", "exec", "do something"], @@ -1106,6 +1114,8 @@ async def test_replay_runner_never_invokes_process_lifecycle_callbacks(tmp_path: env={"SCENARIO_STEP_NAME": "recorded"}, on_process_spawned=on_spawned, on_process_reaped=on_reaped, + rollout=rollout, + logical_run_start_ts="2026-10-10T19:00:00+00:00", ) on_spawned.assert_not_called() diff --git a/tests/execution/test_session_index_roundtrip.py b/tests/execution/test_session_index_roundtrip.py index c1f85baf51..8eb6196790 100644 --- a/tests/execution/test_session_index_roundtrip.py +++ b/tests/execution/test_session_index_roundtrip.py @@ -3,9 +3,11 @@ from __future__ import annotations import json +from datetime import datetime import pytest +from autoskillit.execution.session_log.session_index import iter_tolerant_session_index_lines from tests.execution.conftest import _flush pytestmark = [pytest.mark.layer("execution"), pytest.mark.medium] @@ -38,6 +40,7 @@ def test_appending_current_schema_row_preserves_retained_v8_row_bytes(self, tmp_ index_path = tmp_path / "sessions.jsonl" retained = json.loads(index_path.read_text().strip()) retained["schema_version"] = 8 + retained.pop("rollout") retained.pop("subagent_model_outcomes") retained_line = json.dumps(retained, sort_keys=True, separators=(",", ":")) index_path.write_text(retained_line + "\n") @@ -59,6 +62,51 @@ def test_appending_current_schema_row_preserves_retained_v8_row_bytes(self, tmp_ assert lines[0] == retained_line old_row, new_row = map(json.loads, lines) assert old_row["schema_version"] == 8 + assert "rollout" not in old_row assert "subagent_model_outcomes" not in old_row assert new_row["schema_version"] == SESSION_INDEX_SCHEMA_VERSION assert new_row["subagent_model_outcomes"] == [outcome] + + def test_streaming_reader_classifies_exact_rollout_cutover_boundary(self, tmp_path): + index_path = tmp_path / "sessions.jsonl" + effective_at = "2026-11-01T18:00:00.500000+00:00" + starts = ( + ("before", "2026-11-01T18:00:00.499999+00:00"), + ("equal", "2026-11-01T18:00:00.500000+00:00"), + ("after", "2026-11-01T18:00:00.500001+00:00"), + ) + marker = { + "change_id": "prompt-cost-study", + "effective_at": effective_at, + "scope": {"skills": [], "backends": [], "roles": []}, + "expected_effect": "Reduce planning input tokens.", + } + rows = [{"timestamp": timestamp, "rollout": marker} for _name, timestamp in starts] + rows.extend( + [ + {"timestamp": "2026-11-01T18:00:00.500000+00:00"}, + {"timestamp": "2026-11-01T18:00:00.500000+00:00", "rollout": None}, + ] + ) + index_path.write_text("".join(json.dumps(row) + "\n" for row in rows)) + + cohorts = [] + unassigned = 0 + for _end_offset, row in iter_tolerant_session_index_lines(index_path): + if row is None: + continue + rollout = row.get("rollout") + if rollout is None: + unassigned += 1 + continue + started_at = datetime.fromisoformat(row["timestamp"]) + cutover_at = datetime.fromisoformat(rollout["effective_at"]) + if started_at < cutover_at: + cohorts.append("before") + elif started_at >= cutover_at: + cohorts.append("after") + else: + raise AssertionError("timezone-aware timestamps must have an ordering") + + assert cohorts == ["before", "after", "after"] + assert unassigned == 2 diff --git a/tests/execution/test_session_log_backend_source.py b/tests/execution/test_session_log_backend_source.py index 1660484855..b3b821fd4b 100644 --- a/tests/execution/test_session_log_backend_source.py +++ b/tests/execution/test_session_log_backend_source.py @@ -111,6 +111,7 @@ def test_session_index_entry_default_value(self) -> None: "adjudication_verdict": None, "native_shell_capture": None, "session_type": None, + "rollout": None, "subagent_model_outcomes": [], "child_outcomes": [], "schema_version": SESSION_INDEX_SCHEMA_VERSION, diff --git a/tests/execution/test_session_log_fields.py b/tests/execution/test_session_log_fields.py index ca588bbb56..4558bc6059 100644 --- a/tests/execution/test_session_log_fields.py +++ b/tests/execution/test_session_log_fields.py @@ -21,6 +21,7 @@ SessionType, ) from autoskillit.core.types.results._type_results import ModelIdentity +from tests._helpers import make_rollout_marker from tests.execution.conftest import ( _flush, _make_cc_jsonl_record, @@ -32,6 +33,7 @@ pytestmark = [pytest.mark.layer("execution"), pytest.mark.medium] _SYNTHETIC_MODEL_ID = "gpt-synthetic" +_ROLLOUT_MARKER = make_rollout_marker() class _FakeLocator: @@ -2103,3 +2105,29 @@ def test_missing_evidence_worktree_persists_not_observed_defaults(tmp_path): assert row["evidence_worktree_path"] == "" assert row["evidence_worktree_source"] == "not_observed" assert row["evidence_worktree_detail"] == "" + + +def test_rollout_marker_is_identical_in_summary_and_index(tmp_path): + _flush( + tmp_path, + session_id="marked-rollout", + rollout=_ROLLOUT_MARKER, + proc_snapshots=None, + ) + + summary = json.loads((tmp_path / "sessions" / "marked-rollout" / "summary.json").read_text()) + index = json.loads((tmp_path / "sessions.jsonl").read_text().strip()) + + assert summary["rollout"] == _ROLLOUT_MARKER + assert index["rollout"] == _ROLLOUT_MARKER + assert summary["rollout"] == index["rollout"] + + +def test_inactive_rollout_is_null_in_summary_and_index(tmp_path): + _flush(tmp_path, session_id="unmarked-rollout", proc_snapshots=None) + + summary = json.loads((tmp_path / "sessions" / "unmarked-rollout" / "summary.json").read_text()) + index = json.loads((tmp_path / "sessions.jsonl").read_text().strip()) + + assert summary["rollout"] is None + assert index["rollout"] is None diff --git a/tests/execution/test_session_log_retention.py b/tests/execution/test_session_log_retention.py index 11bb39a506..7355fef8d4 100644 --- a/tests/execution/test_session_log_retention.py +++ b/tests/execution/test_session_log_retention.py @@ -27,6 +27,7 @@ ) from autoskillit.execution.session_log.session_index import read_tolerant_session_index_rows from autoskillit.fleet import FLEET_STATE_SCHEMA_VERSION, build_protected_campaign_ids +from tests._helpers import make_rollout_marker from tests._retention_surface import ( RECLAIMER_CONVERGENCE_CASES, assert_second_pass_is_quiet, @@ -206,11 +207,12 @@ def test_recover_crashed_sessions_finalizes_orphaned_file(tmp_path): """recover_crashed_sessions reads tmpfs file and writes permanent session dir.""" tmpfs = tmp_path / "shm" tmpfs.mkdir() - _write_old_trace( + trace = _write_old_trace( tmpfs, "autoskillit_trace_12345.jsonl", json.dumps({"vm_rss_kb": 500, "captured_at": "2026-03-03T10:00:00+00:00"}) + "\n", ) + expected_start_ts = datetime.fromtimestamp(trace.stat().st_mtime, tz=UTC).isoformat() count = recover_crashed_sessions(tmpfs_path=str(tmpfs), log_dir=str(tmp_path / "logs")) assert count == 1 sessions = list((tmp_path / "logs" / "sessions").iterdir()) @@ -219,6 +221,7 @@ def test_recover_crashed_sessions_finalizes_orphaned_file(tmp_path): assert (sessions[0] / "summary.json").exists() summary = json.loads((sessions[0] / "summary.json").read_text()) assert summary["termination_reason"] == "CRASHED" + assert summary["start_ts"] == expected_start_ts assert summary["success"] is False assert summary["infra_exit_category"] == "unclassified" assert summary["infra_cleanup_incomplete"] is False @@ -228,10 +231,52 @@ def test_recover_crashed_sessions_finalizes_orphaned_file(tmp_path): assert len(index_rows) == 1 index = json.loads(index_rows[0]) assert index["infra_exit_category"] == "unclassified" + assert index["timestamp"] == expected_start_ts assert index["infra_cleanup_incomplete"] is False assert index["infra_fault_domain"] == "infrastructure" +def test_recovery_preserves_interrupted_nudge_rollout_and_logical_start(tmp_path): + tmpfs = tmp_path / "shm" + tmpfs.mkdir() + pid = 99992 + trace = _write_old_trace( + tmpfs, + f"autoskillit_trace_{pid}.jsonl", + json.dumps( + { + "vm_rss_kb": 500, + "captured_at": "2026-03-03T10:00:00+00:00", + "comm": "claude", + } + ) + + "\n", + ) + enrollment = tmpfs / f"autoskillit_enrollment_{pid}.json" + sidecar = json.loads(enrollment.read_text()) + rollout = make_rollout_marker() + logical_run_start_ts = "2000-01-01T00:00:00+00:00" + sidecar.update( + schema_version=3, + comm="claude", + rollout=rollout, + logical_run_start_ts=logical_run_start_ts, + ) + enrollment.write_text(json.dumps(sidecar)) + trace_mtime_ts = datetime.fromtimestamp(trace.stat().st_mtime, tz=UTC).isoformat() + + assert recover_crashed_sessions(tmpfs_path=str(tmpfs), log_dir=str(tmp_path / "logs")) == 1 + + session_dir = next((tmp_path / "logs" / "sessions").iterdir()) + summary = json.loads((session_dir / "summary.json").read_text()) + index = json.loads((tmp_path / "logs" / "sessions.jsonl").read_text().splitlines()[0]) + assert logical_run_start_ts != trace_mtime_ts + assert summary["start_ts"] == logical_run_start_ts + assert index["timestamp"] == logical_run_start_ts + assert summary["rollout"] == rollout + assert index["rollout"] == rollout + + def test_recovery_replay_upserts_without_rewriting_committed_directory( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: diff --git a/tests/infra/test_e2e_workflow.py b/tests/infra/test_e2e_workflow.py index c1fa814d2e..ede505cd35 100644 --- a/tests/infra/test_e2e_workflow.py +++ b/tests/infra/test_e2e_workflow.py @@ -11,6 +11,7 @@ import pytest +from autoskillit.core import MIN_FREE_BYTES_THRESHOLD from autoskillit.core.io import load_yaml pytestmark = [pytest.mark.layer("infra"), pytest.mark.medium] @@ -289,6 +290,17 @@ def test_headless_smoke_exit_trap_removes_exact_container_then_runs_cleanup( assert statuses[key] == value +def test_e2e_test_container_exceeds_pytest_capacity_floor(workflow: dict, tmp_path: Path) -> None: + result, events, _, _ = _run_smoke_workflow_shell(workflow, tmp_path) + + assert result.returncode == 0, result.stderr + launch_args = events[0]["args"] + shm_arg = next(arg for arg in launch_args if arg.startswith("--shm-size=")) + size = re.fullmatch(r"--shm-size=(\d+)g", shm_arg) + assert size is not None + assert int(size[1]) * 1024**3 > MIN_FREE_BYTES_THRESHOLD + + @pytest.mark.parametrize("redact_status", [0, 1]) def test_local_lifecycle_records_redaction_and_fails_when_redaction_fails( tmp_path: Path, redact_status: int @@ -408,7 +420,7 @@ def test_clean_install_uses_the_common_user_image_and_artifacts( run = _inline_smoke_cleanup(_step(workflow, "run", step_id=step_id)["run"]) branch, outside = _model_branch(run) - assert run.count("docker_args=()") == 1 + assert run.count("docker_args=(") == 1 assert '"${docker_args[@]}"' in outside assert "docker run --rm" in outside assert '"$IMAGE"' in outside diff --git a/tests/test_test_filter_core_cascade.py b/tests/test_test_filter_core_cascade.py index 0ccdcb42f1..febbd4f0bc 100644 --- a/tests/test_test_filter_core_cascade.py +++ b/tests/test_test_filter_core_cascade.py @@ -302,7 +302,9 @@ def test_type_results_cascade(self) -> None: "smoke_utils", } ) - assert MODULE_CASCADE_CORE["_type_results_records"] == MODULE_CASCADE_CORE["_type_results"] + assert MODULE_CASCADE_CORE["_type_results_records"] == ( + MODULE_CASCADE_CORE["_type_results"] | {"config"} + ) def test_type_backend_cascade(self) -> None: assert MODULE_CASCADE_CORE["_type_backend"] == frozenset( @@ -794,7 +796,8 @@ def test_type_results_narrow_routing(self, tmp_path: Path, source_path: str) -> dir_names = {p.name for p in result} for pkg in ["core", "execution", "server", "fleet", "migration"]: assert pkg in dir_names, f"_type_results cascade should include {pkg}" - for excluded in ["config", "hooks", "skills", "planner"]: + assert ("config" in dir_names) == source_path.endswith("_type_results_records.py") + for excluded in ["hooks", "skills", "planner"]: assert excluded not in dir_names, ( f"_type_results cascade should not include {excluded}" )