diff --git a/src/autoskillit/core/__init__.pyi b/src/autoskillit/core/__init__.pyi index 2fc711925d..022d54576c 100644 --- a/src/autoskillit/core/__init__.pyi +++ b/src/autoskillit/core/__init__.pyi @@ -964,6 +964,8 @@ from .types import CoverageStatus as CoverageStatus from .types import DatabaseReader as DatabaseReader from .types import DeclaredTruthUnresolved as DeclaredTruthUnresolved from .types import DeclaredTruthUnsupported as DeclaredTruthUnsupported +from .types import DelegatedWriteObservation as DelegatedWriteObservation +from .types import DelegatedWriteTarget as DelegatedWriteTarget from .types import DeliveryOccurrenceId as DeliveryOccurrenceId from .types import DiffAnchorAuthority as DiffAnchorAuthority from .types import DirectInstall as DirectInstall diff --git a/src/autoskillit/core/types/results/__init__.py b/src/autoskillit/core/types/results/__init__.py index 4c6357fee9..1cfb3e1159 100644 --- a/src/autoskillit/core/types/results/__init__.py +++ b/src/autoskillit/core/types/results/__init__.py @@ -12,6 +12,10 @@ from ._type_results_execution import __all__ as _results_execution_all from ._type_token import * # noqa: F403 from ._type_token import __all__ as _token_all +from ._type_write_evidence import * # noqa: F403 +from ._type_write_evidence import __all__ as _write_evidence_all if not TYPE_CHECKING: - __all__ = _figure_spec_all + _results_all + _results_execution_all + _token_all + __all__ = ( + _figure_spec_all + _write_evidence_all + _results_all + _results_execution_all + _token_all + ) diff --git a/src/autoskillit/core/types/results/_type_results.py b/src/autoskillit/core/types/results/_type_results.py index d5dadd76c1..39f9aad68a 100644 --- a/src/autoskillit/core/types/results/_type_results.py +++ b/src/autoskillit/core/types/results/_type_results.py @@ -65,6 +65,7 @@ require_aware_timestamp, ) from ._type_token import TurnTokenEntry +from ._type_write_evidence import DelegatedWriteTarget, WriteEvidence if TYPE_CHECKING: from ..launch._type_launch import ResolvedLaunchContract @@ -105,7 +106,6 @@ "OutcomeInvariantSpec", "WorkspaceOutcomeRecord", "WriteBehaviorSpec", - "WriteEvidence", "FailureRecord", "ProviderOutcome", "PreLaunchReadiness", @@ -247,11 +247,13 @@ class WriteBehaviorSpec: external_effect: ``none``, ``serialized-idempotent``, or ``serialized-unknown-completion`` for effects outside the workspace. + delegated_target: Canonical in-scope target for delegated writes. """ mode: str | None = None expected_when: tuple[str, ...] = () external_effect: str = "none" + delegated_target: DelegatedWriteTarget | None = None def __post_init__(self) -> None: if self.external_effect not in VALID_EXTERNAL_EFFECTS: @@ -495,38 +497,6 @@ class AuditResultOutcome: attempt_id: AuditAttemptId | None = None -@dataclass(frozen=True, slots=True) -class WriteEvidence: - """Bundled write evidence signals — either explicitly constructed or absent.""" - - write_call_count: int - fs_writes_detected: bool - git_writes_detected: bool - file_changes_count: int = 0 - - @classmethod - def none_observed(cls) -> WriteEvidence: - return cls( - write_call_count=0, - fs_writes_detected=False, - git_writes_detected=False, - file_changes_count=0, - ) - - @property - def has_evidence(self) -> bool: - return ( - self.write_call_count >= 1 - or self.fs_writes_detected - or self.git_writes_detected - or self.file_changes_count >= 1 - ) - - @property - def has_implementation_evidence(self) -> bool: - return self.write_call_count >= 1 or self.file_changes_count >= 1 - - @dataclass class SkillResult: """Typed result returned by _build_skill_result and run_headless_core.""" diff --git a/src/autoskillit/core/types/results/_type_write_evidence.py b/src/autoskillit/core/types/results/_type_write_evidence.py new file mode 100644 index 0000000000..7ed447ac10 --- /dev/null +++ b/src/autoskillit/core/types/results/_type_write_evidence.py @@ -0,0 +1,96 @@ +"""Write evidence and delegated-write target contracts.""" + +from __future__ import annotations + +import hashlib +from dataclasses import dataclass +from pathlib import Path +from stat import S_ISREG + +__all__ = ["DelegatedWriteObservation", "DelegatedWriteTarget", "WriteEvidence"] + + +@dataclass(frozen=True, slots=True) +class DelegatedWriteTarget: + """A canonical absolute target path and its invocation content baseline.""" + + path: str + baseline_digest: str | None + + @staticmethod + def _content_digest(path: str) -> str | None: + """Hash file bytes. + + Metadata changes and identical rewrites do not establish a content change. + Unexpected I/O errors propagate rather than becoming an absent baseline. + """ + candidate = Path(path) + try: + if not S_ISREG(candidate.stat().st_mode): + return None + return hashlib.sha256(candidate.read_bytes()).hexdigest() + except (FileNotFoundError, NotADirectoryError): + return None + + @classmethod + def bind(cls, path: str) -> DelegatedWriteTarget: + """Capture the invocation baseline. + + Admission also needs a successful owned tool observation; this is not an + exclusive filesystem lock or proof of the writer's identity. + """ + return cls(path=path, baseline_digest=cls._content_digest(path)) + + def changed(self) -> bool: + current_digest = self._content_digest(self.path) + return current_digest is not None and current_digest != self.baseline_digest + + +@dataclass(frozen=True, slots=True) +class DelegatedWriteObservation: + """A parsed child write-tool observation.""" + + agent_tool_use_id: str + agent_type: str + tool_use_id: str + tool_name: str + file_path: str + succeeded: bool + + +@dataclass(frozen=True, slots=True) +class WriteEvidence: + """Bundled write evidence signals — either explicitly constructed or absent.""" + + write_call_count: int + fs_writes_detected: bool + git_writes_detected: bool + file_changes_count: int = 0 + delegated_write_count: int = 0 + + @classmethod + def none_observed(cls) -> WriteEvidence: + return cls( + write_call_count=0, + fs_writes_detected=False, + git_writes_detected=False, + file_changes_count=0, + ) + + @property + def has_evidence(self) -> bool: + return ( + self.write_call_count >= 1 + or self.fs_writes_detected + or self.git_writes_detected + or self.file_changes_count >= 1 + or self.delegated_write_count >= 1 + ) + + @property + def has_implementation_evidence(self) -> bool: + return ( + self.write_call_count >= 1 + or self.file_changes_count >= 1 + or self.delegated_write_count >= 1 + ) diff --git a/src/autoskillit/execution/headless/_headless_adjudication.py b/src/autoskillit/execution/headless/_headless_adjudication.py index 49aec7cdfd..c12a2efbca 100644 --- a/src/autoskillit/execution/headless/_headless_adjudication.py +++ b/src/autoskillit/execution/headless/_headless_adjudication.py @@ -37,7 +37,6 @@ from autoskillit.execution.headless._headless_evidence import ( _adapt_agent_result, _apply_budget_guard, - _compute_write_evidence, ) from autoskillit.execution.headless._headless_outcome import ( apply_finding_disposition_adjudication, @@ -58,6 +57,10 @@ parse_session_result, ) from autoskillit.execution.session._session_outcome import _compute_success +from autoskillit.execution.session._write_evidence import ( + WriteEvidenceScope, + _compute_write_evidence, +) if TYPE_CHECKING: from autoskillit.core import ( @@ -86,11 +89,7 @@ class _SkillResultContext: provider_used: str write_behavior: WriteBehaviorSpec | None skill_contract: SkillContract | None - cwd: str - fs_writes_detected: bool - git_writes_detected: bool - file_changes: Sequence[str] - write_watch_dirs: Sequence[Path] + evidence_scope: WriteEvidenceScope backend_resume_session_id: str outcome_ledger: WorkspaceOutcomeLedger | None @@ -494,11 +493,7 @@ def _attempt_stall_recovery( provider_used: str, write_behavior: WriteBehaviorSpec | None, skill_contract: SkillContract | None, - cwd: str, - fs_writes_detected: bool, - git_writes_detected: bool, - file_changes: Sequence[str], - write_watch_dirs: Sequence[Path], + evidence_scope: WriteEvidenceScope, backend_resume_session_id: str = "", outcome_ledger: WorkspaceOutcomeLedger | None = None, ) -> tuple[SkillResult | None, ClaudeSessionResult, WriteEvidence, ApiRetryOutcome]: @@ -518,16 +513,7 @@ def _attempt_stall_recovery( backend_resume_session_id=backend_resume_session_id, provider_used=provider_used, ) - evidence = _compute_write_evidence( - session, - fs_writes_detected, - git_writes_detected, - backend, - file_changes=file_changes, - write_watch_dirs=write_watch_dirs, - cwd=cwd, - skill_command=skill_command, - ) + evidence = _compute_write_evidence(session, evidence_scope) api_retry = _build_api_retry_outcome(session) returncode = result.returncode if result.returncode is not None else -1 can_attempt_recovery = ( @@ -566,7 +552,7 @@ def _attempt_stall_recovery( evidence, write_behavior, skill_contract, - cwd, + evidence_scope.cwd, outcome_ledger=outcome_ledger, start_ts=result.start_ts, end_ts=result.end_ts, diff --git a/src/autoskillit/execution/headless/_headless_evidence.py b/src/autoskillit/execution/headless/_headless_evidence.py index 1aa2fe040b..96ea7afffb 100644 --- a/src/autoskillit/execution/headless/_headless_evidence.py +++ b/src/autoskillit/execution/headless/_headless_evidence.py @@ -4,14 +4,11 @@ import dataclasses import json -from collections.abc import Sequence from datetime import UTC, datetime -from pathlib import Path from typing import TYPE_CHECKING, Any from autoskillit.core import ( CODEX_CONTEXT_EXHAUSTION_MARKER, - WORKTREE_SKILLS, AgentSessionResult, ChildOutcomeDict, CliSubtype, @@ -21,8 +18,6 @@ SessionTelemetry, SkillResult, SubagentModelOutcomeDict, - WriteEvidence, - extract_skill_name, get_logger, is_parent_assistant_record, ) @@ -144,72 +139,6 @@ def _adapt_agent_result(agent_result: AgentSessionResult) -> ClaudeSessionResult ) -def _compute_write_evidence( - session: ClaudeSessionResult, - fs_writes_detected: bool, - git_writes_detected: bool, - backend: CodingAgentBackend, - file_changes: Sequence[str] = (), - write_watch_dirs: Sequence[Path] = (), - cwd: str = "", - skill_command: str = "", -) -> WriteEvidence: - write_names = backend.write_tool_names() - - # Worktree implementation evidence must be outside temp; other skills count temp writes. - extracted = extract_skill_name(skill_command) if skill_command else None - is_worktree_dispatch = bool( - extracted and extracted in WORKTREE_SKILLS and cwd and write_watch_dirs - ) - - if is_worktree_dispatch: - resolved_cwd = str(Path(cwd).resolve()) - temp_prefix = resolved_cwd + "/.autoskillit/temp/" - tracked_write_count = 0 - for t in session.tool_uses: - if t.get("name") not in write_names: - continue - tool_id = t.get("id") - if tool_id is not None and tool_id in session.denied_tool_use_ids: - continue - file_path = t.get("file_path", "") - if not file_path: - continue - resolved = str((Path(resolved_cwd) / file_path).resolve()) - if resolved.startswith(temp_prefix): - continue - tracked_write_count += 1 - write_call_count = tracked_write_count - else: - write_call_count = sum( - 1 - for t in session.tool_uses - if t.get("name") in write_names and t.get("id") not in session.denied_tool_use_ids - ) - - if write_call_count == 0 and file_changes: - if is_worktree_dispatch: - resolved_cwd = str(Path(cwd).resolve()) - temp_prefix = resolved_cwd + "/.autoskillit/temp/" - tracked_file_changes = [ - fc - for fc in file_changes - if not str((Path(resolved_cwd) / fc).resolve()).startswith(temp_prefix) - ] - file_changes_count = len(tracked_file_changes) - else: - file_changes_count = len(file_changes) - else: - file_changes_count = 0 - - return WriteEvidence( - write_call_count=write_call_count, - fs_writes_detected=fs_writes_detected, - git_writes_detected=git_writes_detected, - file_changes_count=file_changes_count, - ) - - def _extract_file_changes(stdout: str, backend: CodingAgentBackend) -> list[str]: if backend.capabilities.write_detection_strategy == "tool_names": return [] diff --git a/src/autoskillit/execution/headless/_headless_result.py b/src/autoskillit/execution/headless/_headless_result.py index 69c4e8ebf0..4418771ec4 100644 --- a/src/autoskillit/execution/headless/_headless_result.py +++ b/src/autoskillit/execution/headless/_headless_result.py @@ -49,7 +49,6 @@ from autoskillit.execution.headless._headless_evidence import ( _apply_budget_guard, _capture_failure, - _compute_write_evidence, _extract_file_changes, _stdout_mentions_write_tools, ) @@ -73,6 +72,10 @@ from autoskillit.execution.session._session_outcome import ( _compute_outcome, ) +from autoskillit.execution.session._write_evidence import ( + WriteEvidenceScope, + _compute_write_evidence, +) from autoskillit.hooks import CODEX_AUTO_COMPACTION_DENIED_REASON if TYPE_CHECKING: @@ -168,11 +171,7 @@ def _build_stall_result( provider_used=context.provider_used, write_behavior=context.write_behavior, skill_contract=context.skill_contract, - cwd=context.cwd, - fs_writes_detected=context.fs_writes_detected, - git_writes_detected=context.git_writes_detected, - file_changes=context.file_changes, - write_watch_dirs=context.write_watch_dirs, + evidence_scope=context.evidence_scope, backend_resume_session_id=context.backend_resume_session_id, outcome_ledger=context.outcome_ledger, ) @@ -225,16 +224,7 @@ def _make_controlled_context_exhaustion_result( context: _SkillResultContext, ) -> SkillResult: """Build the terminal result for the correlated Codex compaction veto.""" - evidence = _compute_write_evidence( - session, - context.fs_writes_detected, - context.git_writes_detected, - context.backend, - file_changes=context.file_changes, - write_watch_dirs=context.write_watch_dirs, - cwd=context.cwd, - skill_command=context.skill_command, - ) + evidence = _compute_write_evidence(session, context.evidence_scope) _, needs_retry, retry_reason = _apply_infra_retry_policy( InfraExitCategory.CONTEXT_EXHAUSTED, outcome=SessionOutcome.FAILED, @@ -300,6 +290,16 @@ def _build_skill_result( ) -> SkillResult: """Route SubprocessResult fields into the standard run_skill response.""" file_changes = _extract_file_changes(result.stdout, backend) + evidence_scope = WriteEvidenceScope( + write_tool_names=backend.write_tool_names(), + fs_writes_detected=fs_writes_detected, + git_writes_detected=git_writes_detected, + file_changes=tuple(file_changes), + write_watch_dirs=tuple(write_watch_dirs), + cwd=cwd, + skill_command=skill_command, + delegated_target=getattr(write_behavior, "delegated_target", None), + ) context = _SkillResultContext( backend=backend, completion_marker=completion_marker, @@ -311,11 +311,7 @@ def _build_skill_result( provider_used=provider_used, write_behavior=write_behavior, skill_contract=skill_contract, - cwd=cwd, - fs_writes_detected=fs_writes_detected, - git_writes_detected=git_writes_detected, - file_changes=file_changes, - write_watch_dirs=write_watch_dirs, + evidence_scope=evidence_scope, backend_resume_session_id=backend_resume_session_id, outcome_ledger=outcome_ledger, ) @@ -375,16 +371,7 @@ def _build_skill_result( context=context, ) obligation_session = session - obligation_evidence = _compute_write_evidence( - obligation_session, - fs_writes_detected, - git_writes_detected, - backend, - file_changes=file_changes, - write_watch_dirs=write_watch_dirs, - cwd=cwd, - skill_command=skill_command, - ) + obligation_evidence = _compute_write_evidence(obligation_session, context.evidence_scope) diagnostics = [f"pending={','.join(obligation_pending)}"] if obligation_pending else [] if obligation_wakeup: diagnostics.append("schedule_wakeup=true") @@ -479,16 +466,7 @@ def _build_skill_result( else: returncode = result.returncode if result.returncode is not None else -1 - evidence = _compute_write_evidence( - session, - fs_writes_detected, - git_writes_detected, - backend, - file_changes=file_changes, - write_watch_dirs=write_watch_dirs, - cwd=cwd, - skill_command=skill_command, - ) + evidence = _compute_write_evidence(session, context.evidence_scope) _has_write_evidence = evidence.has_evidence try: @@ -776,7 +754,7 @@ def _build_skill_result( evidence, write_behavior, skill_contract, - cwd, + evidence_scope.cwd, outcome_ledger=outcome_ledger, start_ts=result.start_ts, end_ts=result.end_ts, diff --git a/src/autoskillit/execution/headless/_headless_terminal.py b/src/autoskillit/execution/headless/_headless_terminal.py index df19d770b0..d3934c58f6 100644 --- a/src/autoskillit/execution/headless/_headless_terminal.py +++ b/src/autoskillit/execution/headless/_headless_terminal.py @@ -402,6 +402,7 @@ def build_terminal_flush_kwargs( "fs_writes_detected": skill_result.evidence.fs_writes_detected, "git_writes_detected": skill_result.evidence.git_writes_detected, "file_changes_count": skill_result.evidence.file_changes_count, + "delegated_write_count": skill_result.evidence.delegated_write_count, "clone_contamination_reverted": clone_contamination_reverted, "tracked_comm": result.tracked_comm, "orphaned_tool_result": result.orphaned_tool_result, diff --git a/src/autoskillit/execution/session/_provider_parse.py b/src/autoskillit/execution/session/_provider_parse.py index ecb6aa94b6..3abdaea05a 100644 --- a/src/autoskillit/execution/session/_provider_parse.py +++ b/src/autoskillit/execution/session/_provider_parse.py @@ -13,11 +13,17 @@ import json from collections.abc import Mapping from dataclasses import dataclass, field -from typing import Any, assert_never +from typing import Any, TypeGuard, assert_never import regex as re -from autoskillit.core import CONTEXT_EXHAUSTION_MARKER, ClaudeContentBlockType, get_logger +from autoskillit.core import ( + CONTEXT_EXHAUSTION_MARKER, + ClaudeContentBlockType, + DelegatedWriteObservation, + get_logger, + is_parent_assistant_record, +) logger = get_logger(__name__) @@ -25,6 +31,7 @@ _HANDLED_RECORD_TYPES: frozenset[str] = frozenset( {"assistant", "rate_limit_event", "result", "system", "user"} ) +_AGENT_SPAWN_TOOL_NAMES = frozenset({"Agent", "Task"}) def _is_compaction_status_record(obj: Mapping[str, Any]) -> bool: @@ -67,6 +74,10 @@ class _ProviderParseAccumulator: provider_error_code: str = "" api_error_message_seen: bool = False denied_tool_use_ids: set[str] = field(default_factory=set) + agent_invocations: dict[str, str] = field(default_factory=dict) + child_write_calls: list[dict[str, str]] = field(default_factory=list) + child_tool_errors: dict[tuple[str, str], bool] = field(default_factory=dict) + delegated_writes: tuple[DelegatedWriteObservation, ...] = () def _parse_provider_records(stdout: str) -> _ProviderParseAccumulator: @@ -104,6 +115,7 @@ def _parse_provider_records(stdout: str) -> _ProviderParseAccumulator: _accumulate_provider_record(fallback, acc) except json.JSONDecodeError: pass + acc.delegated_writes = _build_delegated_writes(acc) return acc @@ -162,6 +174,9 @@ def _accumulate_provider_record(obj: dict[str, Any], acc: _ProviderParseAccumula else max(acc.rate_limit_resets_at_epoch, raw_resets_at) ) return + _register_root_agent_invocations(obj, acc) + _capture_child_write_calls(obj, acc) + _capture_child_tool_results(obj, acc) if record_type == "result": acc.result_obj = obj raw_status = _provider_field(obj, "api_error_status") @@ -193,6 +208,164 @@ def _accumulate_provider_record(obj: dict[str, Any], acc: _ProviderParseAccumula acc.denied_tool_use_ids.add(tool_use_id) +def _nonempty_string(value: Any) -> TypeGuard[str]: + return isinstance(value, str) and bool(value) + + +def _record_content_blocks(obj: dict[str, Any]) -> list[Any] | None: + message = obj.get("message") + if not isinstance(message, dict): + return None + content = message.get("content") + if isinstance(content, list): + return content + return None + + +def _child_record_identity(obj: dict[str, Any], record_type: str) -> tuple[str, str] | None: + """Return validated child type and parent tool id for one record.""" + if obj.get("type") != record_type: + return None + subagent_type = obj.get("subagent_type") + parent_tool_use_id = obj.get("parent_tool_use_id") + if not _nonempty_string(subagent_type): + return None + if not _nonempty_string(parent_tool_use_id): + return None + return subagent_type, parent_tool_use_id + + +def _root_agent_invocation(block: Any) -> tuple[str, str] | None: + if not isinstance(block, dict) or block.get("type") != "tool_use": + return None + name = block.get("name") + if not _nonempty_string(name) or name not in _AGENT_SPAWN_TOOL_NAMES: + return None + tool_use_id = block.get("id") + if not _nonempty_string(tool_use_id): + return None + tool_input = block.get("input") + if not isinstance(tool_input, Mapping): + return None + subagent_type = tool_input.get("subagent_type") + if _nonempty_string(subagent_type): + return tool_use_id, subagent_type + return None + + +def _register_root_agent_invocations(obj: dict[str, Any], acc: _ProviderParseAccumulator) -> None: + """Retain requested types from real, top-level Agent and Task calls.""" + if not is_parent_assistant_record(obj): + return + if obj.get("parent_tool_use_id"): + return + content = _record_content_blocks(obj) + if content is None: + return + for block in content: + invocation = _root_agent_invocation(block) + if invocation is None: + continue + tool_use_id, subagent_type = invocation + acc.agent_invocations[tool_use_id] = subagent_type + + +def _child_write_call( + block: Any, subagent_type: str, parent_tool_use_id: str +) -> dict[str, str] | None: + if not isinstance(block, dict) or block.get("type") != "tool_use": + return None + name = block.get("name") + if not _nonempty_string(name) or name not in {"Write", "Edit"}: + return None + tool_use_id = block.get("id") + if not _nonempty_string(tool_use_id): + return None + tool_input = block.get("input") + if not isinstance(tool_input, Mapping): + return None + file_path = tool_input.get("file_path") + if _nonempty_string(file_path): + return { + "id": tool_use_id, + "name": name, + "file_path": file_path, + "parent_tool_use_id": parent_tool_use_id, + "subagent_type": subagent_type, + } + return None + + +def _capture_child_write_calls(obj: dict[str, Any], acc: _ProviderParseAccumulator) -> None: + """Capture child writes using the archive's top-level ``subagent_type`` tag. + + Missing or mismatched tags yield no observation; they are never inferred from + the root invocation. + """ + identity = _child_record_identity(obj, "assistant") + if identity is None: + return + subagent_type, parent_tool_use_id = identity + content = _record_content_blocks(obj) + if content is None: + return + for block in content: + call = _child_write_call(block, subagent_type, parent_tool_use_id) + if call is not None: + acc.child_write_calls.append(call) + + +def _capture_child_tool_results(obj: dict[str, Any], acc: _ProviderParseAccumulator) -> None: + """OR-latch child tool errors under the matching parent invocation.""" + identity = _child_record_identity(obj, "user") + if identity is None: + return + subagent_type, parent_tool_use_id = identity + if acc.agent_invocations.get(parent_tool_use_id) != subagent_type: + return + content = _record_content_blocks(obj) + if content is None: + return + for block in content: + if not isinstance(block, dict) or block.get("type") != "tool_result": + continue + tool_use_id = block.get("tool_use_id") + if not _nonempty_string(tool_use_id): + continue + key = (parent_tool_use_id, tool_use_id) + acc.child_tool_errors[key] = ( + acc.child_tool_errors.get(key, False) or block.get("is_error") is True + ) + + +def _build_delegated_writes( + acc: _ProviderParseAccumulator, +) -> tuple[DelegatedWriteObservation, ...]: + """Build observations owned by an allowed, matching root invocation.""" + observations: list[DelegatedWriteObservation] = [] + for call in acc.child_write_calls: + parent_tool_use_id = call["parent_tool_use_id"] + if parent_tool_use_id not in acc.agent_invocations: + continue + if parent_tool_use_id in acc.denied_tool_use_ids: + continue + agent_type = acc.agent_invocations[parent_tool_use_id] + if call["subagent_type"] != agent_type: + continue + key = (parent_tool_use_id, call["id"]) + observations.append( + DelegatedWriteObservation( + agent_tool_use_id=parent_tool_use_id, + agent_type=agent_type, + tool_use_id=call["id"], + tool_name=call["name"], + file_path=call["file_path"], + succeeded=(key in acc.child_tool_errors and not acc.child_tool_errors[key]), + ) + ) + return tuple(observations) + + def _capture_assistant_record(obj: dict[str, Any], acc: _ProviderParseAccumulator) -> None: """Retain message text, tool uses, and context evidence from an assistant record.""" message = obj.get("message") diff --git a/src/autoskillit/execution/session/_session_model.py b/src/autoskillit/execution/session/_session_model.py index 585a2741bd..20bff122fd 100644 --- a/src/autoskillit/execution/session/_session_model.py +++ b/src/autoskillit/execution/session/_session_model.py @@ -15,6 +15,7 @@ CONTEXT_EXHAUSTION_MARKER, ClaudeContentBlockType, CliSubtype, + DelegatedWriteObservation, RetryReason, SessionOutcome, TokenMeasure, @@ -97,6 +98,7 @@ class ClaudeSessionResult: seen_ndjson_unknown_event_count: int = 0 seen_ndjson_unknown_item_count: int = 0 denied_tool_use_ids: frozenset[str] = field(default_factory=frozenset) + delegated_writes: tuple[DelegatedWriteObservation, ...] = () compaction_attempted: bool = False def __post_init__(self) -> None: @@ -533,4 +535,5 @@ def parse_session_result(stdout: str, *, provider_used: str = "anthropic") -> Cl provider_error_code=acc.provider_error_code, api_error_message_seen=acc.api_error_message_seen, denied_tool_use_ids=frozenset(acc.denied_tool_use_ids), + delegated_writes=acc.delegated_writes, ) diff --git a/src/autoskillit/execution/session/_skill_session_contract_codec.py b/src/autoskillit/execution/session/_skill_session_contract_codec.py index 9cee733d68..f7fb45cfdf 100644 --- a/src/autoskillit/execution/session/_skill_session_contract_codec.py +++ b/src/autoskillit/execution/session/_skill_session_contract_codec.py @@ -21,6 +21,7 @@ SKILL_PROJECTION_VERSION, SKILL_SESSION_CONTRACT_SCHEMA_VERSION, ChildExecutionIdentity, + DelegatedWriteTarget, ExecutionIdentity, ExplorationTaskSpec, ExplorationVectorApplicabilityId, @@ -372,6 +373,27 @@ def _execution_identity_from_dict(value: object) -> ExecutionIdentity: return ExecutionIdentity(**scalar_values, children=tuple(parsed_children)) +def _delegated_target_to_dict(target: DelegatedWriteTarget | None) -> dict[str, Any] | None: + if target is None: + return None + return {"path": target.path, "baseline_digest": target.baseline_digest} + + +def _delegated_target_from_dict(data: Mapping[str, Any]) -> DelegatedWriteTarget | None: + target = data.get("write_behavior", {}).get("delegated_target") + if target is None: + return None + if not isinstance(target, Mapping): + raise ValueError("serialized delegated target must be an object") + path = target["path"] + baseline_digest = target["baseline_digest"] + if not isinstance(path, str) or not path: + raise ValueError("serialized delegated target path must be non-empty text") + if baseline_digest is not None and not isinstance(baseline_digest, str): + raise ValueError("serialized delegated target baseline_digest must be text or null") + return DelegatedWriteTarget(path=path, baseline_digest=baseline_digest) + + def _contract_to_dict(contract: SkillSessionContract) -> dict[str, Any]: exploration_vectors: dict[str, list[object]] = {} for name, vectors in sorted(contract.exploration_vectors.items()): @@ -382,6 +404,7 @@ def _contract_to_dict(contract: SkillSessionContract) -> dict[str, Any]: ): serialized.insert(index, _thaw_json_value(raw)) exploration_vectors[name] = serialized + delegated_target = contract.write_behavior.delegated_target return { "schema_version": contract.schema_version, "root_name": contract.root_name, @@ -430,6 +453,7 @@ def _contract_to_dict(contract: SkillSessionContract) -> dict[str, Any]: "mode": contract.write_behavior.mode, "expected_when": list(contract.write_behavior.expected_when), "external_effect": contract.write_behavior.external_effect, + "delegated_target": _delegated_target_to_dict(delegated_target), }, "read_only": contract.read_only, "scope_discipline": contract.scope_discipline, @@ -650,6 +674,7 @@ def _contract_from_dict(data: Mapping[str, Any]) -> SkillSessionContract: for pattern in data.get("write_behavior", {}).get("expected_when", []) ), external_effect=str(data.get("write_behavior", {}).get("external_effect", "none")), + delegated_target=_delegated_target_from_dict(data), ), read_only=read_only, scope_discipline=scope_discipline, diff --git a/src/autoskillit/execution/session/_write_evidence.py b/src/autoskillit/execution/session/_write_evidence.py new file mode 100644 index 0000000000..68c02fec0b --- /dev/null +++ b/src/autoskillit/execution/session/_write_evidence.py @@ -0,0 +1,168 @@ +"""Write evidence computation for headless sessions.""" + +from __future__ import annotations + +import os +from collections.abc import Collection +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from autoskillit.core import ( + WORKTREE_SKILLS, + DelegatedWriteObservation, + DelegatedWriteTarget, + WriteEvidence, + extract_skill_name, + get_logger, +) +from autoskillit.execution.session._session_model import ClaudeSessionResult + +logger = get_logger(__name__) + + +@dataclass(frozen=True, slots=True) +class WriteEvidenceScope: + write_tool_names: Collection[str] + fs_writes_detected: bool + git_writes_detected: bool + file_changes: tuple[str, ...] + write_watch_dirs: tuple[Path, ...] + cwd: str + skill_command: str + delegated_target: DelegatedWriteTarget | None + + +def _is_worktree_dispatch(scope: WriteEvidenceScope) -> bool: + extracted = extract_skill_name(scope.skill_command) if scope.skill_command else None + return bool( + extracted and extracted in WORKTREE_SKILLS and scope.cwd and scope.write_watch_dirs + ) + + +def _is_implementation_path(scope: WriteEvidenceScope, resolved: str) -> bool: + """Return whether a resolved path is outside the worktree temp directory.""" + if not _is_worktree_dispatch(scope): + return True + temp_prefix = str(Path(scope.cwd).resolve()) + "/.autoskillit/temp/" + return not resolved.startswith(temp_prefix) + + +def _count_worktree_write_calls( + scope: WriteEvidenceScope, + eligible_calls: list[dict[str, Any]], +) -> int: + resolved_cwd = str(Path(scope.cwd).resolve()) + write_call_count = 0 + for tool_use in eligible_calls: + file_path = tool_use.get("file_path", "") + if not file_path: + continue + resolved = str((Path(resolved_cwd) / file_path).resolve()) + if _is_implementation_path(scope, resolved): + write_call_count += 1 + return write_call_count + + +def _count_root_write_calls(session: ClaudeSessionResult, scope: WriteEvidenceScope) -> int: + is_worktree = _is_worktree_dispatch(scope) + eligible_calls: list[dict[str, Any]] = [] + for tool_use in session.tool_uses: + if tool_use.get("name") not in scope.write_tool_names: + continue + tool_id = tool_use.get("id") + if is_worktree: + if tool_id is not None and tool_id in session.denied_tool_use_ids: + continue + elif tool_id in session.denied_tool_use_ids: + continue + eligible_calls.append(tool_use) + + if not is_worktree: + return len(eligible_calls) + return _count_worktree_write_calls(scope, eligible_calls) + + +def _is_admitted_delegated_write( + observation: DelegatedWriteObservation, + target: DelegatedWriteTarget, + scope: WriteEvidenceScope, +) -> bool: + if not observation.succeeded or observation.tool_name not in scope.write_tool_names: + return False + resolved = os.path.realpath(Path(scope.cwd) / observation.file_path) + return resolved == target.path and _is_implementation_path(scope, resolved) + + +def _admitted_delegated_write_count( + session: ClaudeSessionResult, + scope: WriteEvidenceScope, +) -> int: + observations = session.delegated_writes + if not observations: + return 0 + + target = scope.delegated_target + target_bound = target is not None + try: + target_changed = target.changed() if target is not None else False + except OSError as exc: + logger.warning( + "delegated_writes_not_admitted", + observed=len(observations), + target_bound=target_bound, + target_changed=False, + error=str(exc), + exc_info=True, + ) + return 0 + if target is not None and target_changed: + admitted = sum( + 1 + for observation in observations + if _is_admitted_delegated_write(observation, target, scope) + ) + else: + admitted = 0 + + if admitted == 0: + logger.warning( + "delegated_writes_not_admitted", + observed=len(observations), + target_bound=target_bound, + target_changed=target_changed, + ) + return admitted + + +def _compute_write_evidence( + session: ClaudeSessionResult, + scope: WriteEvidenceScope, +) -> WriteEvidence: + # Worktree implementation evidence must be outside temp; other skills count temp writes. + write_call_count = _count_root_write_calls(session, scope) + delegated_write_count = _admitted_delegated_write_count(session, scope) + + if write_call_count == 0 and scope.file_changes: + if _is_worktree_dispatch(scope): + resolved_cwd = str(Path(scope.cwd).resolve()) + file_changes_count = sum( + 1 + for file_change in scope.file_changes + if _is_implementation_path( + scope, + str((Path(resolved_cwd) / file_change).resolve()), + ) + ) + else: + file_changes_count = len(scope.file_changes) + else: + file_changes_count = 0 + + return WriteEvidence( + write_call_count=write_call_count, + fs_writes_detected=scope.fs_writes_detected, + git_writes_detected=scope.git_writes_detected, + file_changes_count=file_changes_count, + delegated_write_count=delegated_write_count, + ) diff --git a/src/autoskillit/execution/session_log/session_log.py b/src/autoskillit/execution/session_log/session_log.py index ee46fc73a8..5ca9d051e5 100644 --- a/src/autoskillit/execution/session_log/session_log.py +++ b/src/autoskillit/execution/session_log/session_log.py @@ -175,6 +175,7 @@ def flush_session_log( fs_writes_detected: bool = False, git_writes_detected: bool = False, file_changes_count: int = 0, + delegated_write_count: int = 0, clone_contamination_reverted: bool = False, tracked_comm: str | None = None, comm_aliases: frozenset[str] = frozenset(), @@ -456,6 +457,7 @@ def flush_session_log( "fs_writes_detected": fs_writes_detected, "git_writes_detected": git_writes_detected, "file_changes_count": file_changes_count, + "delegated_write_count": delegated_write_count, "clone_contamination_reverted": clone_contamination_reverted, # Tracer target resolution fields (issue #806) "tracked_comm": _effective_tracked_comm, diff --git a/src/autoskillit/recipe/contracts/_contracts_manifest.py b/src/autoskillit/recipe/contracts/_contracts_manifest.py index 3b4ec0323c..365b3886ee 100644 --- a/src/autoskillit/recipe/contracts/_contracts_manifest.py +++ b/src/autoskillit/recipe/contracts/_contracts_manifest.py @@ -257,6 +257,25 @@ def _as_input_preflight(value: object) -> tuple[str, ...]: return tuple(value) +def _parse_delegated_write_input( + skill_name: str, + skill_data: dict[str, Any], + inputs: tuple[SkillInput, ...], + write_behavior: object, +) -> str | None: + if "delegated_write_input" not in skill_data: + return None + value = skill_data["delegated_write_input"] + error_prefix = f"delegated_write_input for skill '{skill_name}'" + if not isinstance(value, str): + raise ValueError(f"{error_prefix} must be a string") + if not any(item.name == value and item.type == "file_path" for item in inputs): + raise ValueError(f"{error_prefix} must name a declared file_path input") + if write_behavior not in ("always", "conditional"): + raise ValueError(f"{error_prefix} requires write_behavior 'always' or 'conditional'") + return value + + def get_skill_contract(skill_name: str, manifest: dict[str, Any]) -> SkillContract | None: """Look up a skill in the manifest and return a SkillContract.""" skills = manifest.get("skills", {}) @@ -276,6 +295,9 @@ def get_skill_contract(skill_name: str, manifest: dict[str, Any]) -> SkillContra examples = skill_data.get("pattern_examples", []) write_behavior = skill_data.get("write_behavior") write_expected_when = skill_data.get("write_expected_when", []) + delegated_write_input = _parse_delegated_write_input( + skill_name, skill_data, inputs, write_behavior + ) external_effect = skill_data.get("external_effect", "none") if not isinstance(external_effect, str): raise ValueError(f"external_effect for skill '{skill_name}' must be a string") @@ -313,6 +335,7 @@ def get_skill_contract(skill_name: str, manifest: dict[str, Any]) -> SkillContra expected_output_patterns=patterns, pattern_examples=examples, write_behavior=write_behavior, + delegated_write_input=delegated_write_input, write_expected_when=write_expected_when, external_effect=external_effect, read_only=read_only, diff --git a/src/autoskillit/recipe/contracts/_contracts_types.py b/src/autoskillit/recipe/contracts/_contracts_types.py index fe221f7cbd..d8ab8f66c3 100644 --- a/src/autoskillit/recipe/contracts/_contracts_types.py +++ b/src/autoskillit/recipe/contracts/_contracts_types.py @@ -133,6 +133,7 @@ class SkillContract: default_factory=dict ) audit_output_mode: AuditOutputMode | None = None + delegated_write_input: str | None = None def __post_init__(self) -> None: if self.external_effect not in VALID_EXTERNAL_EFFECTS: diff --git a/src/autoskillit/recipe/skill_contracts.yaml b/src/autoskillit/recipe/skill_contracts.yaml index 5cc8e6b66c..156ac61ec0 100644 --- a/src/autoskillit/recipe/skill_contracts.yaml +++ b/src/autoskillit/recipe/skill_contracts.yaml @@ -311,6 +311,7 @@ skills: absence_value: "" outputs: [] write_behavior: always + delegated_write_input: plan_path input_preflight: [audit_cycle_inventory, plan_set_coverage] implement-worktree: inputs: diff --git a/src/autoskillit/server/recipe/_recipe_execution.py b/src/autoskillit/server/recipe/_recipe_execution.py index a1ef660cfc..df73102088 100644 --- a/src/autoskillit/server/recipe/_recipe_execution.py +++ b/src/autoskillit/server/recipe/_recipe_execution.py @@ -16,6 +16,7 @@ ArtifactRef, AuditCycleVerifier, AuditVerdict, + BindingFailureCode, BindingMode, BoundScalar, InstalledRecipeExecution, @@ -72,7 +73,7 @@ "bind_attested_runtime_invocation", "build_bound_child_prompt", "build_recipe_execution_snapshot", - "build_standalone_child_prompt", + "build_standalone_child_invocation", "clear_recipe_execution", "complete_audit_finalization_effects", "get_recipe_execution", @@ -711,34 +712,22 @@ def build_bound_child_prompt( ) -def build_standalone_child_prompt( +def build_standalone_child_invocation( skill_command: str, cwd: str, skill_inputs: Mapping[str, str | int | float | bool] | None, *, audit_output_mode: AuditOutputMode | None = None, -) -> str: - """Validate standalone inputs without requiring attested recipe state. +) -> tuple[str, tuple[tuple[str, BoundScalar], ...]]: + """Bind a standalone command and return its prompt with canonical inputs. - An unknown skill with no structured inputs may still be resolved from a - project/plugin installation later in the normal dispatch path. Structured - inputs require a canonical contract because their names and order otherwise - cannot be validated safely. + Unknown skills without structured inputs retain the existing pass-through + behavior. Known skills bind positional and structured inputs through the + same canonical contract. """ - if skill_inputs is None: - if audit_output_mode is not None: - return build_bound_child_prompt( - skill_command, - (), - None, - audit_output_mode=audit_output_mode, - ) - return skill_command - with_args: dict[str, object] = { - "skill_command": skill_command, - "cwd": cwd, - "skill_inputs": skill_inputs, - } + with_args: dict[str, object] = {"skill_command": skill_command, "cwd": cwd} + if skill_inputs is not None: + with_args["skill_inputs"] = skill_inputs binding = bind_step_invocation( "standalone", RecipeStep( @@ -751,15 +740,21 @@ def build_standalone_child_prompt( ) if binding.failures: failure = binding.failures[0] + if skill_inputs is None and failure.code is BindingFailureCode.UNKNOWN_SKILL: + return skill_command, () raise RecipeExecutionAdmissionError( f"standalone_{failure.code.value}", failure.message, ) - return build_bound_child_prompt( - skill_command, - binding.canonical_child_invocation, - None, - audit_output_mode=audit_output_mode, + canonical_inputs = binding.canonical_child_invocation + return ( + build_bound_child_prompt( + skill_command, + canonical_inputs, + None, + audit_output_mode=audit_output_mode, + ), + canonical_inputs, ) diff --git a/src/autoskillit/server/tools/_execution_helpers/_skill_contract.py b/src/autoskillit/server/tools/_execution_helpers/_skill_contract.py index 1fb8af9258..22bbe6c420 100644 --- a/src/autoskillit/server/tools/_execution_helpers/_skill_contract.py +++ b/src/autoskillit/server/tools/_execution_helpers/_skill_contract.py @@ -257,6 +257,7 @@ def deserialize_skill_contract(payload: str) -> SkillContract | None: expected_output_patterns=list(data.get("expected_output_patterns", [])), pattern_examples=list(data.get("pattern_examples", [])), write_behavior=data.get("write_behavior"), + delegated_write_input=data.get("delegated_write_input"), write_expected_when=list(data.get("write_expected_when", [])), external_effect=str(data.get("external_effect", "none")), read_only=read_only, diff --git a/src/autoskillit/server/tools/tools_execution/_run_skill_admission.py b/src/autoskillit/server/tools/tools_execution/_run_skill_admission.py index c523d4ad2f..cfa58664e0 100644 --- a/src/autoskillit/server/tools/tools_execution/_run_skill_admission.py +++ b/src/autoskillit/server/tools/tools_execution/_run_skill_admission.py @@ -43,7 +43,7 @@ RecipeExecutionAdmissionError, bind_attested_runtime_invocation, build_bound_child_prompt, - build_standalone_child_prompt, + build_standalone_child_invocation, get_recipe_execution, record_runtime_binding_digest, resolve_attested_input_preflight, @@ -516,11 +516,13 @@ def _admit_dynamic_recipe_invocation(state: _RunSkillDispatchState) -> str | Non ) if not state.resume_session_id: try: - state.child_skill_command = build_standalone_child_prompt( - state.skill_command, - state.cwd, - state.skill_inputs, - audit_output_mode=state._audit_output_mode, + state.child_skill_command, state._bound_recipe_inputs = ( + build_standalone_child_invocation( + state.skill_command, + state.cwd, + state.skill_inputs, + audit_output_mode=state._audit_output_mode, + ) ) except RecipeExecutionAdmissionError as exc: return _recipe_execution_deny(exc.code, str(exc)) @@ -674,11 +676,13 @@ def _admit_standalone_invocation(state: _RunSkillDispatchState) -> str | None: ) if not state.resume_session_id: try: - state.child_skill_command = build_standalone_child_prompt( - state.skill_command, - state.cwd, - state.skill_inputs, - audit_output_mode=state._audit_output_mode, + state.child_skill_command, state._bound_recipe_inputs = ( + build_standalone_child_invocation( + state.skill_command, + state.cwd, + state.skill_inputs, + audit_output_mode=state._audit_output_mode, + ) ) except RecipeExecutionAdmissionError as exc: return _recipe_execution_deny(exc.code, str(exc)) diff --git a/src/autoskillit/server/tools/tools_execution/_run_skill_session.py b/src/autoskillit/server/tools/tools_execution/_run_skill_session.py index bf42d72535..9211de5a3c 100644 --- a/src/autoskillit/server/tools/tools_execution/_run_skill_session.py +++ b/src/autoskillit/server/tools/tools_execution/_run_skill_session.py @@ -20,12 +20,15 @@ LAUNCH_ID_ENV_VAR, SKILL_COMMAND_DISPLAY_MAX, WORKTREE_SKILLS, + ContainmentError, + DelegatedWriteTarget, SkillContractError, SkillExecutionRole, SkillResult, WriteBehaviorSpec, extract_skill_name, get_logger, + resolve_contained_path, ) from autoskillit.core import current_order_id as _current_order_id from autoskillit.core import current_step_name as _current_step_name @@ -368,6 +371,7 @@ async def _prepare_owned_dispatch_session( if (terminal := _extend_closure_write_scope(state)) is not None: return terminal + _bind_delegated_target(state) _prepare_owned_session_contract(state) if (terminal := _prepare_owned_write_scope(state)) is not None: @@ -527,6 +531,40 @@ def _prepare_owned_write_scope(state: _RunSkillDispatchState) -> str | None: return None +def _bind_delegated_target(state: _RunSkillDispatchState) -> None: + """Bind a declared child write to the canonical in-scope invocation input.""" + input_name = getattr(state._skill_contract, "delegated_write_input", None) + if state._stored_contract is not None or state.write_spec is None or not input_name: + return + + value = dict(state._bound_recipe_inputs).get(input_name) + if value is None: + reason = "input_absent" + elif not isinstance(value, str): + reason = "input_not_string" + elif not value: + reason = "input_empty" + else: + try: + resolved = resolve_contained_path(Path(state.cwd) / value, state.cwd) + target = DelegatedWriteTarget.bind(str(resolved)) + except (ContainmentError, OSError, ValueError) as exc: + reason = str(exc) + else: + state.write_spec = replace( + state.write_spec, + delegated_target=target, + ) + return + + logger.warning( + "delegated_write_target_unbound", + skill=state.skill_command, + input=input_name, + reason=reason, + ) + + def _prepare_owned_session_contract(state: _RunSkillDispatchState) -> None: assert state.skill_add_dirs is not None assert state.resolved_command is not None diff --git a/tests/_test_filter.py b/tests/_test_filter.py index 11fa05cc8c..3300492a91 100644 --- a/tests/_test_filter.py +++ b/tests/_test_filter.py @@ -553,6 +553,7 @@ class SourceMapRejection(enum.StrEnum): "_type_checkpoint": frozenset({"core", "execution", "fleet", "server"}), "_type_results": _TYPE_RESULTS_CASCADE | {"config", "report"}, "_type_results_records": _TYPE_RESULTS_CASCADE | {"config", "report"}, + "_type_write_evidence": _TYPE_RESULTS_CASCADE | {"config", "report"}, "_type_results_execution": frozenset({"core", "execution", "server", "pipeline"}), "_type_execution_identity": frozenset( {"core", "execution", "server", "cli", "fleet", "smoke_utils"} @@ -1210,6 +1211,7 @@ class SourceMapRejection(enum.StrEnum): # autoskillit.pipeline at call time. Validated by REQ-GUARD-007. "execution/test_backend_dispatch.py", "execution/test_boundary_pty_dispatch.py", + "execution/test_delegated_write_evidence.py", "execution/test_flush_provider_integration.py", "execution/test_headless_add_dirs.py", "execution/test_headless_backend_mixing.py", @@ -1335,6 +1337,7 @@ class SourceMapRejection(enum.StrEnum): "migration", # Server file-level entries importing autoskillit.recipe: "server/test_run_skill_execution_tuning_fallbacks.py", + "server/test_run_skill_delegated_write_binding.py", "server/test_bundled_dispatch_write_scope_admission.py", "server/test_serve_idempotence.py", "server/test_attestation_delivery_reachability.py", diff --git a/tests/arch/test_python_no_hardcoded_temp.py b/tests/arch/test_python_no_hardcoded_temp.py index 62a57801d8..94b4c5996f 100644 --- a/tests/arch/test_python_no_hardcoded_temp.py +++ b/tests/arch/test_python_no_hardcoded_temp.py @@ -77,7 +77,7 @@ # writes by constructing the temp prefix for path-aware filtering. The literal mirrors # the canonical default used by resolve_temp_dir — no resolve_temp_dir() available at # this stage because path filtering runs on session.tool_uses captured before session ends. - "execution/headless/_headless_evidence.py": ( + "execution/session/_write_evidence.py": ( "path-aware write-evidence discriminator; mirrors canonical default" ), # Justification: docstring for resolve_state_root() explains that callers append diff --git a/tests/arch/test_subpackage_structure.py b/tests/arch/test_subpackage_structure.py index 71804c66ef..02c1fe5166 100644 --- a/tests/arch/test_subpackage_structure.py +++ b/tests/arch/test_subpackage_structure.py @@ -77,6 +77,7 @@ "_type_results", "_type_results_execution", "_type_results_records", + "_type_write_evidence", "_type_token", "_type_figure_spec", } @@ -353,6 +354,7 @@ def test_session_has_expected_modules(self): "_retry_fsm", "_exit_classification", "_provider_parse", + "_write_evidence", "turn_usage", "_managed_headless_session_lineage", "_managed_headless_session_lineage_codec", diff --git a/tests/contracts/test_skill_md_no_has_implementation_progress_token.py b/tests/contracts/test_skill_md_no_has_implementation_progress_token.py index 699e63df6a..7342f6e082 100644 --- a/tests/contracts/test_skill_md_no_has_implementation_progress_token.py +++ b/tests/contracts/test_skill_md_no_has_implementation_progress_token.py @@ -1,7 +1,8 @@ """Contract: has_implementation_progress must NOT appear as an emit line in any SKILL.md. -The token is server-computed via SkillResult.has_implementation_progress -(src/autoskillit/core/types/results/_type_results.py:526) — it is never parsed from +The token is server-computed via SkillResult.has_implementation_progress, which delegates +to WriteEvidence.has_implementation_evidence +(src/autoskillit/core/types/results/_type_write_evidence.py) — it is never parsed from model text. Including it as an emit line in SKILL.md misleads models into self-reporting progress as authoritative. This test prevents re-introduction. """ diff --git a/tests/contracts/test_skill_result_persistence_ledger.py b/tests/contracts/test_skill_result_persistence_ledger.py index 2ba19ad33c..8badcf894b 100644 --- a/tests/contracts/test_skill_result_persistence_ledger.py +++ b/tests/contracts/test_skill_result_persistence_ledger.py @@ -56,6 +56,7 @@ ("completion_required", "completion_required", "live-only:derived-elsewhere"), ("contamination.retry_reason", "contamination_retry_reason", "live-only:derived-elsewhere"), ("contamination.subtype", "contamination_subtype", "live-only:derived-elsewhere"), + ("evidence.delegated_write_count", "delegated_write_count", "summary-only"), ("evidence.file_changes_count", "file_changes_count", "persisted"), ("evidence.fs_writes_detected", "fs_writes_detected", "persisted"), ("evidence.git_writes_detected", "git_writes_detected", "persisted"), @@ -260,6 +261,7 @@ def test_durable_ledger_rows_exist_in_real_flushed_artifacts(tmp_path) -> None: "api_retry_last_error": "", "api_retry_last_status": None, "cli_subtype": "", + "delegated_write_count": 0, "file_changes_count": 0, "fs_writes_detected": False, "git_writes_detected": False, diff --git a/tests/core/types/test_type_results.py b/tests/core/types/test_type_results.py index c15cb1ceac..4db12add62 100644 --- a/tests/core/types/test_type_results.py +++ b/tests/core/types/test_type_results.py @@ -2,9 +2,11 @@ from __future__ import annotations +from pathlib import Path + import pytest -from autoskillit.core.types import WriteEvidence +from autoskillit.core import DelegatedWriteTarget, WriteEvidence pytestmark = [pytest.mark.layer("core"), pytest.mark.small] @@ -16,6 +18,7 @@ def test_none_observed_all_zero(self) -> None: assert we.fs_writes_detected is False assert we.git_writes_detected is False assert we.file_changes_count == 0 + assert we.delegated_write_count == 0 def test_has_evidence_true_from_write_call_count(self) -> None: we = WriteEvidence(write_call_count=3, fs_writes_detected=False, git_writes_detected=False) @@ -46,6 +49,17 @@ def test_has_evidence_true_from_file_changes_count(self) -> None: ) assert we.has_evidence is True + def test_delegated_write_count_is_implementation_evidence(self) -> None: + ev = WriteEvidence( + write_call_count=0, + fs_writes_detected=False, + git_writes_detected=False, + delegated_write_count=1, + ) + + assert ev.has_evidence is True + assert ev.has_implementation_evidence is True + def test_none_observed_includes_file_changes_count_zero(self) -> None: we = WriteEvidence.none_observed() assert we.file_changes_count == 0 @@ -84,6 +98,51 @@ def test_has_implementation_evidence_true_with_file_changes(self) -> None: assert ev.has_implementation_evidence is True +class TestDelegatedWriteTarget: + @pytest.mark.parametrize("operation", ["stat", "read_bytes"]) + def test_bind_propagates_unexpected_io_errors( + self, tmp_path: Path, monkeypatch, operation: str + ) -> None: + path = tmp_path / "plan.md" + path.write_bytes(b"plan") + + def unreadable(*args, **kwargs): + raise PermissionError("target unavailable") + + with monkeypatch.context() as patch: + patch.setattr(Path, operation, unreadable) + with pytest.raises(PermissionError, match="target unavailable"): + DelegatedWriteTarget.bind(str(path)) + + @pytest.mark.parametrize("is_directory", [False, True], ids=["missing", "directory"]) + def test_bind_records_no_baseline_for_non_files( + self, tmp_path: Path, is_directory: bool + ) -> None: + path = tmp_path / "target" + if is_directory: + path.mkdir() + + target = DelegatedWriteTarget.bind(str(path.resolve())) + + assert target.baseline_digest is None + + def test_changed_tracks_content_and_ignores_identical_bytes_or_deletion( + self, tmp_path: Path + ) -> None: + path = tmp_path / "plan.md" + path.write_bytes(b"before") + target = DelegatedWriteTarget.bind(str(path.resolve())) + + path.write_bytes(b"before") + assert target.changed() is False + + path.write_bytes(b"after") + assert target.changed() is True + + path.unlink() + assert target.changed() is False + + class TestValidateRolloutMarker: def test_normalizes_utc_timestamp_and_copies_scope(self) -> None: from autoskillit.core import validate_rollout_marker diff --git a/tests/execution/conftest.py b/tests/execution/conftest.py index 83315773d6..8ef3152c01 100644 --- a/tests/execution/conftest.py +++ b/tests/execution/conftest.py @@ -37,6 +37,7 @@ from autoskillit.execution.runtime.launch_resolution import DefaultLaunchResolver from autoskillit.execution.session import ClaudeSessionResult from autoskillit.execution.session._exit_classification import _CODEX_API_ERROR_PATTERNS +from autoskillit.execution.session._write_evidence import WriteEvidenceScope from tests._helpers import make_tracing_config from tests.fixtures.codex import FLAT_ERROR_MODEL_CAPACITY, TURN_FAILED_MODEL_CAPACITY @@ -522,6 +523,159 @@ def _subagent_assistant_ndjson( ) +def delegated_root_agent_call( + tool_use_id: str | None = "call_agent_A", + agent_type: str = "general-purpose", +) -> str: + block: dict[str, Any] = { + "type": "tool_use", + "name": "Agent", + "input": { + "subagent_type": agent_type, + "model": "sonnet", + "description": "Update the plan", + "prompt": "Read and update the plan.", + }, + } + if tool_use_id is not None: + block["id"] = tool_use_id + return json.dumps( + { + "type": "assistant", + "parent_tool_use_id": None, + "message": {"model": "claude-sonnet-4-6", "content": [block]}, + } + ) + + +def delegated_task_started(tool_use_id: str = "call_agent_A") -> str: + return json.dumps({"type": "system", "subtype": "task_started", "tool_use_id": tool_use_id}) + + +def delegated_task_notification(tool_use_id: str = "call_agent_A") -> str: + return json.dumps( + {"type": "system", "subtype": "task_notification", "tool_use_id": tool_use_id} + ) + + +def delegated_child_prompt( + parent_tool_use_id: str = "call_agent_A", + agent_type: str = "general-purpose", +) -> str: + return json.dumps( + { + "type": "user", + "parent_tool_use_id": parent_tool_use_id, + "subagent_type": agent_type, + "message": {"role": "user", "content": "Read and update the plan."}, + } + ) + + +def delegated_child_tool_use( + tool_use_id: Any = "call_child_edit", + tool_name: str = "Edit", + file_path: Any = "/workspace/plan.md", + parent_tool_use_id: Any = "call_agent_A", + agent_type: Any = "general-purpose", +) -> str: + block: dict[str, Any] = {"type": "tool_use", "name": tool_name, "id": tool_use_id} + if file_path is not None: + block["input"] = {"file_path": file_path} + record: dict[str, Any] = {"type": "assistant", "message": {"content": [block]}} + if parent_tool_use_id is not None: + record["parent_tool_use_id"] = parent_tool_use_id + if agent_type is not None: + record["subagent_type"] = agent_type + return json.dumps(record) + + +def delegated_child_tool_result( + tool_use_id: Any = "call_child_edit", + is_error: bool | None = None, + parent_tool_use_id: Any = "call_agent_A", + agent_type: Any = "general-purpose", +) -> str: + block: dict[str, Any] = {"type": "tool_result", "tool_use_id": tool_use_id} + if is_error is not None: + block["is_error"] = is_error + record: dict[str, Any] = { + "type": "user", + "message": {"role": "user", "content": [block]}, + } + if parent_tool_use_id is not None: + record["parent_tool_use_id"] = parent_tool_use_id + if agent_type is not None: + record["subagent_type"] = agent_type + return json.dumps(record) + + +def delegated_root_agent_result() -> str: + return json.dumps( + { + "type": "user", + "message": { + "role": "user", + "content": [ + { + "type": "tool_result", + "tool_use_id": "call_agent_A", + "content": [{"type": "text", "text": "Plan updated."}], + } + ], + }, + "tool_use_result": {"agentId": "agentA", "status": "completed"}, + } + ) + + +def delegated_root_agent_text_result() -> str: + return json.dumps( + { + "type": "user", + "message": { + "role": "user", + "content": [ + { + "type": "tool_result", + "tool_use_id": "call_agent_A", + "content": "Plan updated.\nagentId: agentA", + } + ], + }, + } + ) + + +def delegated_root_text(text: str = "Plan updated.") -> str: + return json.dumps( + {"type": "assistant", "message": {"content": [{"type": "text", "text": text}]}} + ) + + +def delegated_result_record() -> str: + return _result_ndjson("Plan updated.") + + +def delegated_stream(records: list[str] | tuple[str, ...]) -> str: + return "\n".join(records) + + +def make_write_evidence_scope(**overrides: Any) -> WriteEvidenceScope: + values = { + "write_tool_names": {"Write", "Edit"}, + "fs_writes_detected": False, + "git_writes_detected": False, + "file_changes": (), + "write_watch_dirs": (), + "cwd": "", + "skill_command": "", + "delegated_target": None, + } + values.update(overrides) + return WriteEvidenceScope(**values) + + def _flush( tmp_path: Path, *, backend: str = "claude-code", session_locator=None, **overrides ) -> None: diff --git a/tests/execution/headless/test_codex_attempt_retention.py b/tests/execution/headless/test_codex_attempt_retention.py index 31e3ac8064..3532ea0266 100644 --- a/tests/execution/headless/test_codex_attempt_retention.py +++ b/tests/execution/headless/test_codex_attempt_retention.py @@ -23,7 +23,7 @@ SkillResult, ) from autoskillit.core.types import KillReason, SubprocessResult, TerminationReason -from autoskillit.core.types.results._type_results import WriteEvidence +from autoskillit.core.types.results._type_write_evidence import WriteEvidence 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 diff --git a/tests/execution/headless/test_headless_evidence_split_integrity.py b/tests/execution/headless/test_headless_evidence_split_integrity.py index 67e791f4a8..0f7f877d90 100644 --- a/tests/execution/headless/test_headless_evidence_split_integrity.py +++ b/tests/execution/headless/test_headless_evidence_split_integrity.py @@ -1,6 +1,6 @@ -"""Split integrity tests for execution/headless/ _headless_evidence split. +"""Split integrity tests for headless evidence and session write evidence. -Verifies that symbols moved to _headless_evidence are importable. +Verifies that symbols moved across those modules remain importable. """ import pytest @@ -12,7 +12,7 @@ class TestHeadlessEvidenceModuleExists: - """Symbols moved to _headless_evidence are importable from there.""" + """Evidence symbols remain importable from their owning modules.""" def test__adapt_agent_result_importable(self): from autoskillit.execution.headless._headless_evidence import _adapt_agent_result @@ -20,7 +20,7 @@ def test__adapt_agent_result_importable(self): assert callable(_adapt_agent_result) def test__compute_write_evidence_importable(self): - from autoskillit.execution.headless._headless_evidence import _compute_write_evidence + from autoskillit.execution.session._write_evidence import _compute_write_evidence assert callable(_compute_write_evidence) diff --git a/tests/execution/headless/test_headless_recovery.py b/tests/execution/headless/test_headless_recovery.py index b395e7107a..f90f618e51 100644 --- a/tests/execution/headless/test_headless_recovery.py +++ b/tests/execution/headless/test_headless_recovery.py @@ -24,7 +24,7 @@ SubprocessResult, TerminationReason, ) -from autoskillit.core.types.results._type_results import WriteEvidence +from autoskillit.core.types.results._type_write_evidence 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, _unmarked_rollout_snapshot diff --git a/tests/execution/test_delegated_write_evidence.py b/tests/execution/test_delegated_write_evidence.py new file mode 100644 index 0000000000..9344e5daf2 --- /dev/null +++ b/tests/execution/test_delegated_write_evidence.py @@ -0,0 +1,464 @@ +"""Delegated child writes can satisfy a bound write target without polluting root evidence.""" + +from __future__ import annotations + +import json +import os +from dataclasses import replace +from pathlib import Path +from types import SimpleNamespace + +import pytest +from structlog.testing import capture_logs + +from autoskillit.core import DelegatedWriteTarget, RetryReason, WriteBehaviorSpec +from autoskillit.core.types import TerminationReason +from autoskillit.execution.backends.claude import ClaudeCodeBackend +from autoskillit.execution.headless import _build_skill_result, run_headless_core +from autoskillit.execution.headless._headless_adjudication import ( + _apply_contract_recovery_gate, +) +from autoskillit.execution.headless._headless_recovery import ( + _infer_enum_token_from_write_contract, + _synthesize_from_write_artifacts, +) +from autoskillit.execution.session import parse_session_result +from tests.conftest import _make_result +from tests.execution.conftest import ( + _mock_backend, + _result_ndjson, + _sr, + delegated_child_prompt, + delegated_child_tool_result, + delegated_child_tool_use, + delegated_result_record, + delegated_root_agent_call, + delegated_root_agent_result, + delegated_root_agent_text_result, + delegated_root_text, + delegated_stream, + delegated_task_notification, + delegated_task_started, +) + +pytestmark = [pytest.mark.layer("execution"), pytest.mark.small] + +_BASE_PLAN = "Plan draft\n" +_PREPENDED_MARKER = "Dry-walkthrough verified = TRUE\n" + + +@pytest.fixture +def plan_and_target(tmp_path: Path) -> tuple[Path, DelegatedWriteTarget]: + plan_dir = tmp_path / ".autoskillit" / "temp" / "make-plan" + plan_dir.mkdir(parents=True) + plan = plan_dir / "plan.md" + plan.write_text(_BASE_PLAN) + return plan, DelegatedWriteTarget.bind(os.path.realpath(plan)) + + +def _edit_stream(file_path: str, result_record: str | None = None) -> str: + return delegated_stream( + [ + delegated_root_agent_call(), + delegated_task_started(), + delegated_child_prompt(), + delegated_child_tool_use(file_path=file_path), + delegated_child_tool_result(), + delegated_task_notification(), + delegated_root_agent_result(), + delegated_root_text(), + result_record or delegated_result_record(), + ] + ) + + +def _bound_behavior(target: DelegatedWriteTarget | None) -> WriteBehaviorSpec: + return WriteBehaviorSpec(mode="always", delegated_target=target) + + +@pytest.mark.parametrize("fs_writes_detected", [False, True]) +def test_bound_child_edit_is_implementation_progress( + tmp_path: Path, + plan_and_target: tuple[Path, DelegatedWriteTarget], + fs_writes_detected: bool, +) -> None: + plan, target = plan_and_target + plan.write_text(_PREPENDED_MARKER + plan.read_text()) + result = _build_skill_result( + _make_result(returncode=0, stdout=_edit_stream(str(plan))), + skill_command=f"/autoskillit:dry-walkthrough {plan}", + cwd=str(tmp_path), + write_watch_dirs=(plan.parent,), + write_behavior=_bound_behavior(target), + fs_writes_detected=fs_writes_detected, + backend=ClaudeCodeBackend(), + ) + + assert result.success is True + assert result.subtype != "zero_writes" + assert result.evidence.write_call_count == 0 + assert result.evidence.file_changes_count == 0 + assert result.evidence.delegated_write_count == 1 + assert result.has_progress_evidence is True + assert result.has_implementation_progress is True + + +def test_unreadable_target_rejects_child_write_and_logs_cause( + tmp_path: Path, + plan_and_target: tuple[Path, DelegatedWriteTarget], + monkeypatch, +) -> None: + plan, target = plan_and_target + + def unreadable(path: Path) -> bytes: + raise PermissionError("target unavailable") + + monkeypatch.setattr(Path, "read_bytes", unreadable) + with capture_logs() as events: + result = _build_skill_result( + _make_result(returncode=0, stdout=_edit_stream(str(plan))), + skill_command=f"/autoskillit:dry-walkthrough {plan}", + cwd=str(tmp_path), + write_behavior=_bound_behavior(target), + backend=ClaudeCodeBackend(), + ) + + assert result.evidence.delegated_write_count == 0 + assert any( + event.get("event") == "delegated_writes_not_admitted" + and event.get("error") == "target unavailable" + and event.get("exc_info") is True + for event in events + ) + + +def test_text_handback_without_task_lifecycle_records_is_admitted( + tmp_path: Path, + plan_and_target: tuple[Path, DelegatedWriteTarget], +) -> None: + plan, target = plan_and_target + plan.write_text(_PREPENDED_MARKER + plan.read_text()) + stdout = delegated_stream( + [ + delegated_root_agent_call(), + delegated_child_prompt(), + delegated_child_tool_use(file_path=str(plan)), + delegated_child_tool_result(), + delegated_root_agent_text_result(), + delegated_root_text(), + delegated_result_record(), + ] + ) + + result = _build_skill_result( + _make_result(returncode=0, stdout=stdout), + skill_command=f"/autoskillit:dry-walkthrough {plan}", + cwd=str(tmp_path), + write_behavior=_bound_behavior(target), + backend=ClaudeCodeBackend(), + ) + + assert result.success is True + assert result.subtype != "zero_writes" + assert result.evidence.delegated_write_count == 1 + + +def test_root_edit_remains_root_write_evidence(tmp_path: Path) -> None: + root_edit = json.dumps( + { + "type": "assistant", + "message": { + "content": [ + { + "type": "tool_use", + "id": "root_edit", + "name": "Edit", + "input": {"file_path": str(tmp_path / "root.md")}, + } + ] + }, + } + ) + result = _build_skill_result( + _make_result( + returncode=0, stdout=delegated_stream([root_edit, delegated_result_record()]) + ), + skill_command="/autoskillit:dry-walkthrough root.md", + cwd=str(tmp_path), + write_behavior=_bound_behavior(None), + backend=ClaudeCodeBackend(), + ) + + assert result.evidence.write_call_count == 1 + assert result.evidence.delegated_write_count == 0 + + +@pytest.mark.parametrize( + ("scenario", "tool_name", "parent_id", "is_error", "mutation", "fs_writes", "bound"), + [ + ("child-read", "Read", "call_agent_A", False, "none", False, True), + ("unchanged-bytes", "Write", "call_agent_A", False, "none", False, True), + ("denied-edit", "Edit", "call_agent_A", True, "plan", False, True), + ("unowned-edit", "Edit", "call_other", False, "plan", False, True), + ("sibling-edit", "Edit", "call_agent_A", False, "sibling", False, True), + ("filesystem-noise", None, "call_agent_A", False, "none", True, True), + ("unbound-target", "Edit", "call_agent_A", False, "plan", False, False), + ], +) +def test_non_admitted_child_activity_still_triggers_zero_writes( + tmp_path: Path, + plan_and_target: tuple[Path, DelegatedWriteTarget], + scenario: str, + tool_name: str | None, + parent_id: str, + is_error: bool, + mutation: str, + fs_writes: bool, + bound: bool, +) -> None: + plan, target = plan_and_target + file_path = plan + if scenario == "unchanged-bytes": + plan.write_text(_PREPENDED_MARKER + plan.read_text()) + target = DelegatedWriteTarget.bind(os.path.realpath(plan)) + plan.write_text(plan.read_text()) + elif mutation == "plan": + plan.write_text(_PREPENDED_MARKER + plan.read_text()) + elif mutation == "sibling": + file_path = tmp_path / "sibling.md" + file_path.write_text("changed sibling\n") + + records = [delegated_root_agent_call(), delegated_task_started(), delegated_child_prompt()] + if tool_name is not None: + records.extend( + [ + delegated_child_tool_use( + tool_use_id=f"child_{scenario}", + tool_name=tool_name, + file_path=str(file_path), + parent_tool_use_id=parent_id, + ), + delegated_child_tool_result( + tool_use_id=f"child_{scenario}", + is_error=True if is_error else None, + parent_tool_use_id=parent_id, + ), + ] + ) + records.extend( + [delegated_task_notification(), delegated_root_agent_result(), delegated_result_record()] + ) + stdout = delegated_stream(records) + behavior = _bound_behavior(target if bound else None) + + if bound: + result = _build_skill_result( + _make_result(returncode=0, stdout=stdout), + skill_command=f"/autoskillit:dry-walkthrough {plan}", + cwd=str(tmp_path), + write_behavior=behavior, + fs_writes_detected=fs_writes, + backend=ClaudeCodeBackend(), + ) + else: + with capture_logs() as captured: + result = _build_skill_result( + _make_result(returncode=0, stdout=stdout), + skill_command=f"/autoskillit:dry-walkthrough {plan}", + cwd=str(tmp_path), + write_behavior=behavior, + fs_writes_detected=fs_writes, + backend=ClaudeCodeBackend(), + ) + assert any(log.get("event") == "delegated_writes_not_admitted" for log in captured) + + assert result.subtype == "zero_writes" + assert result.evidence.delegated_write_count == 0 + + +def test_worktree_temp_target_is_not_admitted(tmp_path: Path) -> None: + plan_dir = tmp_path / ".autoskillit" / "temp" / "implement-worktree-no-merge" + plan_dir.mkdir(parents=True) + plan = plan_dir / "plan.md" + plan.write_text(_BASE_PLAN) + target = DelegatedWriteTarget.bind(os.path.realpath(plan)) + plan.write_text(_PREPENDED_MARKER + plan.read_text()) + + result = _build_skill_result( + _make_result(returncode=0, stdout=_edit_stream(str(plan))), + skill_command=f"/autoskillit:implement-worktree-no-merge {plan}", + cwd=str(tmp_path), + write_watch_dirs=(plan_dir,), + write_behavior=_bound_behavior(target), + backend=ClaudeCodeBackend(), + ) + + assert result.subtype == "zero_writes" + assert result.evidence.delegated_write_count == 0 + + +def test_worktree_tracked_target_is_admitted(tmp_path: Path) -> None: + plan = tmp_path / "src" / "x.py" + plan.parent.mkdir(parents=True) + plan.write_text("before\n") + target = DelegatedWriteTarget.bind(os.path.realpath(plan)) + plan.write_text("after\n") + + result = _build_skill_result( + _make_result(returncode=0, stdout=_edit_stream(str(plan))), + skill_command=f"/autoskillit:implement-worktree-no-merge {plan}", + cwd=str(tmp_path), + write_watch_dirs=(tmp_path / ".autoskillit" / "temp",), + write_behavior=_bound_behavior(target), + backend=ClaudeCodeBackend(), + ) + + assert result.success is True + assert result.evidence.delegated_write_count == 1 + + +@pytest.mark.parametrize("recovery_path", ["adjudicated_failure", "empty_output"]) +def test_delegated_only_evidence_reaches_completion_recovery_gates( + tmp_path: Path, + plan_and_target: tuple[Path, DelegatedWriteTarget], + recovery_path: str, +) -> None: + plan, target = plan_and_target + plan.write_text(_PREPENDED_MARKER + plan.read_text()) + output_record = delegated_result_record() + if recovery_path == "empty_output": + output_record = json.dumps( + { + "type": "result", + "subtype": "empty_output", + "is_error": True, + "result": "", + "session_id": "test-sess", + } + ) + stdout = _edit_stream(str(plan), result_record=output_record) + result = _build_skill_result( + _make_result(returncode=0, stdout=stdout), + skill_command=f"/autoskillit:dry-walkthrough {plan}", + cwd=str(tmp_path), + write_behavior=_bound_behavior(target), + backend=ClaudeCodeBackend(), + ) + assert result.evidence.delegated_write_count == 1 + + if recovery_path == "adjudicated_failure": + adjudicated = replace( + result, + success=False, + needs_retry=False, + subtype="adjudicated_failure", + retry_reason=RetryReason.NONE, + ) + promoted = _apply_contract_recovery_gate( + adjudicated, + has_write_evidence=result.evidence.has_evidence, + readonly_skill=False, + returncode=0, + skill_command=f"/autoskillit:dry-walkthrough {plan}", + audit=None, + max_consecutive_retries=3, + ) + assert promoted.needs_retry is True + assert promoted.retry_reason is RetryReason.CONTRACT_RECOVERY + else: + assert result.retry_reason is RetryReason.COMPLETED_NO_FLUSH + + session = parse_session_result(_edit_stream(str(plan))) + expected_patterns = (r"^plan_path\s*=\s*/.+", r"^verdict\s*=\s*real_fix$") + assert ( + _synthesize_from_write_artifacts( + session, + list(expected_patterns), + write_call_count=0, + file_changes=(), + ) + is None + ) + contract = SimpleNamespace( + write_behavior="conditional", + write_expected_when=(r"verdict\s*=\s*real_fix",), + outputs=(SimpleNamespace(name="verdict", allowed_values=("real_fix",)),), + ) + assert ( + _infer_enum_token_from_write_contract( + session, + expected_patterns, + contract, + write_call_count=0, + file_changes=(), + ) + is None + ) + + +def test_stale_success_recovery_keeps_delegated_write_evidence( + tmp_path: Path, + plan_and_target: tuple[Path, DelegatedWriteTarget], +) -> None: + plan, target = plan_and_target + plan.write_text(_PREPENDED_MARKER + plan.read_text()) + result = _build_skill_result( + _sr( + 0, + _edit_stream(str(plan)), + termination=TerminationReason.STALE, + ), + skill_command=f"/autoskillit:dry-walkthrough {plan}", + cwd=str(tmp_path), + write_behavior=_bound_behavior(target), + backend=ClaudeCodeBackend(), + ) + + assert result.success is True + assert result.evidence.delegated_write_count == 1 + + +@pytest.mark.anyio +@pytest.mark.parametrize("write_marker", [True, False], ids=["child-edits", "no-op-retry"]) +async def test_run_headless_core_composes_delegated_write_evidence( + tmp_path: Path, + minimal_ctx, + monkeypatch, + write_marker: bool, +) -> None: + plan_dir = tmp_path / ".autoskillit" / "temp" / "make-plan" + plan_dir.mkdir(parents=True) + plan = plan_dir / "plan.md" + plan.write_text((_PREPENDED_MARKER if not write_marker else "") + _BASE_PLAN) + target = DelegatedWriteTarget.bind(os.path.realpath(plan)) + + async def runner(cmd, **kwargs): + if cmd and cmd[0] == "git": + return _make_result(returncode=1, stdout="") + if write_marker: + plan.write_text(_PREPENDED_MARKER + plan.read_text()) + return _make_result( + returncode=0, + stdout=_edit_stream(str(plan), _result_ndjson("Plan updated.\n%%ORDER_UP%%")), + ) + + monkeypatch.setattr(minimal_ctx, "runner", runner) + monkeypatch.setattr( + minimal_ctx, + "backend", + _mock_backend(pty_required=True, channel_b_capable=True), + ) + result = await run_headless_core( + f"/autoskillit:dry-walkthrough {plan}", + str(tmp_path), + minimal_ctx, + write_watch_dirs=(plan_dir,), + write_behavior=_bound_behavior(target), + ) + + if write_marker: + assert result.success is True + assert result.evidence.delegated_write_count == 1 + else: + assert result.subtype == "zero_writes" + assert result.evidence.delegated_write_count == 0 diff --git a/tests/execution/test_headless_result.py b/tests/execution/test_headless_result.py index 55e98940aa..10187e56c5 100644 --- a/tests/execution/test_headless_result.py +++ b/tests/execution/test_headless_result.py @@ -36,18 +36,19 @@ from autoskillit.execution.headless._headless_adjudication import _parse_stdout from autoskillit.execution.headless._headless_evidence import ( _adapt_agent_result, - _compute_write_evidence, _stdout_mentions_write_tools, ) from autoskillit.execution.session import ClaudeSessionResult from autoskillit.execution.session._exit_classification import classify_infra_exit from autoskillit.execution.session._session_outcome import _compute_outcome +from autoskillit.execution.session._write_evidence import _compute_write_evidence from autoskillit.execution.session.turn_usage import build_turn_token_entry from tests.execution.conftest import ( CODEX_OBSERVED_PROVIDER_FAILURE_CASES, _make_tool_use_line, _sr, _success_session_json, + make_write_evidence_scope, ) from tests.fixtures.codex import ( HAPPY_PATH_SINGLE_TURN, @@ -997,14 +998,12 @@ def test_single_file_change_has_evidence(self): session_id="s1", tool_uses=[], ) - backend = Mock() - backend.write_tool_names.return_value = frozenset() evidence = _compute_write_evidence( session, - False, - False, - backend=backend, - file_changes=["src/foo.py"], + make_write_evidence_scope( + write_tool_names=frozenset(), + file_changes=("src/foo.py",), + ), ) assert evidence.has_evidence is True assert evidence.file_changes_count == 1 @@ -1018,14 +1017,12 @@ def test_multiple_file_changes_has_evidence(self): session_id="s1", tool_uses=[], ) - backend = Mock() - backend.write_tool_names.return_value = frozenset() evidence = _compute_write_evidence( session, - False, - False, - backend=backend, - file_changes=["a.py", "b.py", "c.py"], + make_write_evidence_scope( + write_tool_names=frozenset(), + file_changes=("a.py", "b.py", "c.py"), + ), ) assert evidence.has_evidence is True assert evidence.file_changes_count == 3 @@ -1039,14 +1036,12 @@ def test_empty_file_changes_no_write_tools_no_evidence(self): session_id="s1", tool_uses=[], ) - backend = Mock() - backend.write_tool_names.return_value = frozenset() evidence = _compute_write_evidence( session, - False, - False, - backend=backend, - file_changes=[], + make_write_evidence_scope( + write_tool_names=frozenset(), + file_changes=(), + ), ) assert evidence.has_evidence is False assert evidence.file_changes_count == 0 @@ -2083,7 +2078,11 @@ def test_compute_write_evidence_sets_file_changes_count_when_no_write_calls(self tool_uses=[], ) evidence = _compute_write_evidence( - session, False, False, backend=ClaudeCodeBackend(), file_changes=["a.py", "b.py"] + session, + make_write_evidence_scope( + write_tool_names=ClaudeCodeBackend().write_tool_names(), + file_changes=("a.py", "b.py"), + ), ) assert evidence.file_changes_count == 2 assert evidence.has_evidence is True @@ -2098,7 +2097,11 @@ def test_compute_write_evidence_ignores_file_changes_when_write_calls_exist(self tool_uses=[{"name": "Write", "id": "t1"}], ) evidence = _compute_write_evidence( - session, False, False, backend=ClaudeCodeBackend(), file_changes=["a.py"] + session, + make_write_evidence_scope( + write_tool_names=ClaudeCodeBackend().write_tool_names(), + file_changes=("a.py",), + ), ) assert evidence.file_changes_count == 0 assert evidence.write_call_count == 1 diff --git a/tests/execution/test_outcome_invariants.py b/tests/execution/test_outcome_invariants.py index 0573d259be..f258b28268 100644 --- a/tests/execution/test_outcome_invariants.py +++ b/tests/execution/test_outcome_invariants.py @@ -19,11 +19,8 @@ from autoskillit.core import RetryReason, WriteBehaviorSpec from autoskillit.core.types import KillReason -from autoskillit.core.types.results._type_results import ( - ApiRetryOutcome, - SkillResult, - WriteEvidence, -) +from autoskillit.core.types.results._type_results import ApiRetryOutcome, SkillResult +from autoskillit.core.types.results._type_write_evidence import WriteEvidence from autoskillit.execution.backends.claude import ClaudeCodeBackend from autoskillit.execution.headless import _build_skill_result from autoskillit.execution.headless._headless_adjudication import ( diff --git a/tests/execution/test_session_log_fields.py b/tests/execution/test_session_log_fields.py index 951930d53b..9b5ce7e04c 100644 --- a/tests/execution/test_session_log_fields.py +++ b/tests/execution/test_session_log_fields.py @@ -199,6 +199,20 @@ def test_flush_session_log_write_call_count_defaults_to_zero(tmp_path): assert summary["write_call_count"] == 0 +def test_flush_session_log_delegated_write_count_is_summary_only(tmp_path): + _flush( + tmp_path, + session_id="delegated-writes", + delegated_write_count=3, + proc_snapshots=None, + ) + summary = json.loads((tmp_path / "sessions" / "delegated-writes" / "summary.json").read_text()) + index = json.loads((tmp_path / "sessions.jsonl").read_text().strip().split("\n")[-1]) + + assert summary["delegated_write_count"] == 3 + assert "delegated_write_count" not in index + + def test_flush_session_log_writes_kitchen_id(tmp_path): """kitchen_id parameter is written to sessions.jsonl index entry.""" _flush( diff --git a/tests/execution/test_session_parsing.py b/tests/execution/test_session_parsing.py index 46f11cf2c7..50aa8f53ff 100644 --- a/tests/execution/test_session_parsing.py +++ b/tests/execution/test_session_parsing.py @@ -8,6 +8,7 @@ import pytest import structlog +from autoskillit.core import DelegatedWriteObservation from autoskillit.core.types import ( ChannelConfirmation, InfraExitCategory, @@ -25,7 +26,20 @@ from tests._helpers import UNKNOWN_MEASURE as _UNKNOWN from tests._helpers import _flush_structlog_proxy_caches as _flush_logger_proxy_caches from tests._helpers import observed_measure as _observed -from tests.execution.conftest import _assistant_ndjson, _result_ndjson +from tests.execution.conftest import ( + _assistant_ndjson, + _result_ndjson, + delegated_child_prompt, + delegated_child_tool_result, + delegated_child_tool_use, + delegated_result_record, + delegated_root_agent_call, + delegated_root_agent_result, + delegated_root_text, + delegated_stream, + delegated_task_notification, + delegated_task_started, +) pytestmark = [pytest.mark.layer("execution"), pytest.mark.small] @@ -1425,6 +1439,326 @@ def test_parse_session_result_excludes_subagent_assistant_messages(self): assert "subagent says goodbye" not in " ".join(result.assistant_messages) +def _delegated_case_stream(root_call: str, *records: str) -> str: + return delegated_stream( + [root_call, *records, delegated_root_agent_result(), delegated_result_record()] + ) + + +class TestDelegatedWriteObservations: + def test_archive_shaped_child_edit_preserves_root_view(self) -> None: + root_read = json.dumps( + { + "type": "assistant", + "message": { + "content": [ + { + "type": "tool_use", + "id": "call_root_read", + "name": "Read", + "input": {"file_path": "/workspace/README.md"}, + } + ] + }, + } + ) + stream = delegated_stream( + [ + json.dumps({"type": "system", "subtype": "init", "session_id": "root"}), + root_read, + delegated_root_agent_call(), + delegated_task_started(), + delegated_child_prompt(), + delegated_child_tool_use( + tool_use_id="call_child_read", + tool_name="Read", + file_path="/workspace/README.md", + ), + delegated_child_tool_result(tool_use_id="call_child_read"), + delegated_child_tool_use( + file_path="/workspace/.autoskillit/temp/make-plan/plan.md" + ), + delegated_child_tool_result(), + delegated_child_tool_use( + tool_use_id="call_child_readback", + tool_name="Read", + file_path="/workspace/.autoskillit/temp/make-plan/plan.md", + ), + delegated_child_tool_result(tool_use_id="call_child_readback"), + delegated_task_notification(), + delegated_root_agent_result(), + delegated_root_text(), + delegated_result_record(), + ] + ) + result = parse_session_result(stream) + root_only = delegated_stream( + [line for line in stream.splitlines() if "subagent_type" not in json.loads(line)] + ) + root_result = parse_session_result(root_only) + + assert result.tool_uses == root_result.tool_uses + assert result.denied_tool_use_ids == root_result.denied_tool_use_ids + assert result.delegated_writes == ( + DelegatedWriteObservation( + agent_tool_use_id="call_agent_A", + agent_type="general-purpose", + tool_use_id="call_child_edit", + tool_name="Edit", + file_path="/workspace/.autoskillit/temp/make-plan/plan.md", + succeeded=True, + ), + ) + + @pytest.mark.parametrize( + "child_result", + [delegated_child_tool_result(is_error=True), None], + ids=["denied", "missing-result"], + ) + def test_child_edit_is_unsuccessful_when_result_errors_or_is_missing( + self, child_result: str | None + ) -> None: + records = [delegated_child_tool_use(file_path="/workspace/plan.md")] + if child_result is not None: + records.append(child_result) + result = parse_session_result( + _delegated_case_stream(delegated_root_agent_call(), *records) + ) + assert len(result.delegated_writes) == 1 + assert result.delegated_writes[0].succeeded is False + + @pytest.mark.parametrize( + "child_records", + [ + [ + delegated_child_tool_use(parent_tool_use_id="call_not_issued_by_root"), + delegated_child_tool_result(parent_tool_use_id="call_not_issued_by_root"), + ], + [ + json.dumps( + { + "type": "assistant", + "parent_tool_use_id": "call_agent_A", + "subagent_type": "general-purpose", + "message": { + "content": [ + { + "type": "tool_use", + "id": "call_nested_agent", + "name": "Agent", + "input": {"subagent_type": "Explore"}, + } + ] + }, + } + ), + delegated_child_tool_use( + tool_use_id="nested_edit", + parent_tool_use_id="call_nested_agent", + agent_type="Explore", + ), + delegated_child_tool_result( + tool_use_id="nested_edit", + parent_tool_use_id="call_nested_agent", + agent_type="Explore", + ), + ], + [ + json.dumps( + { + "type": "assistant", + "parent_tool_use_id": "call_agent_A", + "message": { + "content": [ + { + "type": "tool_use", + "id": "call_untagged_agent", + "name": "Agent", + "input": {"subagent_type": "Explore"}, + } + ] + }, + } + ), + delegated_child_tool_use( + tool_use_id="nested_edit", + parent_tool_use_id="call_untagged_agent", + agent_type="Explore", + ), + delegated_child_tool_result( + tool_use_id="nested_edit", + parent_tool_use_id="call_untagged_agent", + agent_type="Explore", + ), + ], + [ + delegated_child_tool_use(parent_tool_use_id=None), + delegated_child_tool_result(parent_tool_use_id=None), + ], + [ + delegated_child_tool_use(agent_type=None), + delegated_child_tool_result(agent_type=None), + ], + [ + delegated_child_tool_use(agent_type="Explore"), + delegated_child_tool_result(agent_type="Explore"), + ], + [ + json.dumps( + { + "type": "user", + "message": { + "content": [ + { + "type": "tool_result", + "tool_use_id": "call_agent_A", + "is_error": True, + } + ] + }, + } + ), + delegated_child_tool_use(), + delegated_child_tool_result(), + ], + ], + ids=[ + "unowned-parent", + "nested-agent-call", + "untagged-nested-agent-call", + "missing-parent", + "untagged-write", + "tag-mismatch", + "denied-root-agent-call", + ], + ) + def test_rejects_unowned_or_mismatched_child_records(self, child_records: list[str]) -> None: + result = parse_session_result( + _delegated_case_stream(delegated_root_agent_call(), *child_records) + ) + assert result.delegated_writes == () + + def test_synthetic_root_agent_record_does_not_own_child_write(self) -> None: + synthetic_root = json.dumps( + { + "type": "assistant", + "message": { + "model": "", + "content": [ + { + "type": "tool_use", + "id": "call_agent_A", + "name": "Agent", + "input": {"subagent_type": "general-purpose"}, + } + ], + }, + } + ) + result = parse_session_result( + _delegated_case_stream( + synthetic_root, + delegated_child_tool_use(), + delegated_child_tool_result(), + ) + ) + assert result.delegated_writes == () + + @pytest.mark.parametrize( + "stream", + [ + _delegated_case_stream( + delegated_root_agent_call(tool_use_id=None), + delegated_child_tool_use(parent_tool_use_id=None), + delegated_child_tool_result(parent_tool_use_id=None), + ), + _delegated_case_stream( + delegated_root_agent_call(tool_use_id="123"), + delegated_child_tool_use(parent_tool_use_id=123), + delegated_child_tool_result(parent_tool_use_id=123), + ), + _delegated_case_stream( + delegated_root_agent_call(agent_type="123"), + delegated_child_tool_use(agent_type=123), + delegated_child_tool_result(agent_type=123), + ), + ], + ids=["missing-root-id", "integer-parent-id", "integer-child-tag"], + ) + def test_rejects_missing_or_non_string_ownership_identity(self, stream: str) -> None: + result = parse_session_result(stream) + assert result.delegated_writes == () + + @pytest.mark.parametrize( + ("tool_use_id", "child_result"), + [ + ("123", delegated_child_tool_result(tool_use_id=123)), + ("call_child_edit", delegated_child_tool_result(agent_type="Explore")), + ], + ids=["integer-result-id", "mismatched-result-tag"], + ) + def test_malformed_or_mismatched_result_cannot_mark_write_successful( + self, tool_use_id: str, child_result: str + ) -> None: + result = parse_session_result( + _delegated_case_stream( + delegated_root_agent_call(), + delegated_child_tool_use(tool_use_id=tool_use_id), + child_result, + ) + ) + assert len(result.delegated_writes) == 1 + assert result.delegated_writes[0].succeeded is False + + def test_child_results_are_scoped_when_sibling_tool_ids_collide(self) -> None: + stream = delegated_stream( + [ + delegated_root_agent_call(), + delegated_root_agent_call(tool_use_id="call_agent_B", agent_type="Explore"), + delegated_child_tool_use( + tool_use_id="shared_child_edit", + parent_tool_use_id="call_agent_A", + agent_type="general-purpose", + ), + delegated_child_tool_use( + tool_use_id="shared_child_edit", + parent_tool_use_id="call_agent_B", + agent_type="Explore", + ), + delegated_child_tool_result( + tool_use_id="shared_child_edit", + parent_tool_use_id="call_agent_A", + agent_type="general-purpose", + ), + delegated_result_record(), + ] + ) + + observations = parse_session_result(stream).delegated_writes + + assert [(item.agent_tool_use_id, item.succeeded) for item in observations] == [ + ("call_agent_A", True), + ("call_agent_B", False), + ] + + @pytest.mark.parametrize( + "child_results", + [ + [delegated_child_tool_result(is_error=True), delegated_child_tool_result()], + [delegated_child_tool_result(), delegated_child_tool_result(is_error=True)], + ], + ids=["error-then-success", "success-then-error"], + ) + def test_duplicate_child_result_errors_are_latched(self, child_results: list[str]) -> None: + result = parse_session_result( + _delegated_case_stream( + delegated_root_agent_call(), delegated_child_tool_use(), *child_results + ) + ) + assert len(result.delegated_writes) == 1 + assert result.delegated_writes[0].succeeded is False + + @pytest.mark.parametrize( ("backend", "provider", "field", "value", "expected"), [ diff --git a/tests/execution/test_skill_session_contract_store.py b/tests/execution/test_skill_session_contract_store.py index 7f665b1e5e..bebd6bb2f6 100644 --- a/tests/execution/test_skill_session_contract_store.py +++ b/tests/execution/test_skill_session_contract_store.py @@ -205,6 +205,95 @@ def test_store_round_trip_preserves_machine_contract_and_projected_snapshot( assert stored.snapshot_dir.resolve().is_relative_to(root.resolve()) +@pytest.mark.parametrize("bind_target", [True, False], ids=["bound-target", "no-target"]) +def test_store_round_trip_preserves_delegated_target_baseline( + tmp_path: Path, bind_target: bool +) -> None: + from autoskillit.core import DelegatedWriteTarget, WriteBehaviorSpec + from autoskillit.execution.session import DefaultSkillSessionContractStore + + plan_path = tmp_path / ".autoskillit" / "temp" / "make-plan" / "plan.md" + target: DelegatedWriteTarget | None = None + if bind_target: + plan_path.parent.mkdir(parents=True) + plan_path.write_bytes(b"original plan") + target = DelegatedWriteTarget.bind(str(plan_path.resolve())) + + write_behavior = WriteBehaviorSpec(mode="always", delegated_target=target) + contract = replace( + _contract(tmp_path, "projected\n"), + write_behavior=write_behavior, + ) + store = DefaultSkillSessionContractStore(root=tmp_path / "contracts") + correlation_key = store.create_provisional( + contract=contract, + snapshot={".claude/skills/root/SKILL.md": "projected\n"}, + ) + store.observe_candidate(correlation_key, "provider-attempt-1") + store.finalize(correlation_key, "backend/session:delegated-write") + + if target is not None: + plan_path.write_bytes(b"corrected plan") + + restored = store.load("backend/session:delegated-write").contract.write_behavior + restored_target = restored.delegated_target + assert restored == write_behavior + if target is None: + assert restored_target is None + else: + assert restored_target is not None + assert restored_target.baseline_digest == target.baseline_digest + assert restored_target.changed() is True + + +def test_contract_codec_defaults_absent_delegated_target_under_schema_five( + tmp_path: Path, +) -> None: + from autoskillit.core import WriteBehaviorSpec + from autoskillit.execution.session._skill_session_contract_codec import ( + _contract_from_dict, + _contract_to_dict, + ) + + contract = replace( + _contract(tmp_path, "projected\n"), + write_behavior=WriteBehaviorSpec(mode="always", delegated_target=None), + ) + encoded = _contract_to_dict(contract) + assert encoded["schema_version"] == 5 + encoded["write_behavior"].pop("delegated_target") + + restored = _contract_from_dict(encoded) + + assert restored.schema_version == 5 + assert restored.write_behavior.delegated_target is None + + +@pytest.mark.parametrize( + "target", + [ + [], + {"baseline_digest": None}, + {"path": "plan.md"}, + {"path": "", "baseline_digest": None}, + {"path": 42, "baseline_digest": None}, + {"path": "plan.md", "baseline_digest": 42}, + ], + ids=["not-object", "missing-path", "missing-digest", "empty-path", "path-type", "digest-type"], +) +def test_contract_codec_rejects_malformed_delegated_target(tmp_path: Path, target: object) -> None: + from autoskillit.execution.session._skill_session_contract_codec import ( + _contract_from_dict, + _contract_to_dict, + ) + + encoded = _contract_to_dict(_contract(tmp_path, "projected\n")) + encoded["write_behavior"]["delegated_target"] = target + + with pytest.raises(ValueError, match="Invalid serialized"): + _contract_from_dict(encoded) + + def test_store_persists_lineage_binding_separately_and_rebinds_verified_final_id( tmp_path: Path, ) -> None: diff --git a/tests/execution/test_zero_write_detection.py b/tests/execution/test_zero_write_detection.py index ad0996dbda..c23b1ad5a6 100644 --- a/tests/execution/test_zero_write_detection.py +++ b/tests/execution/test_zero_write_detection.py @@ -8,14 +8,32 @@ from __future__ import annotations import json +import os from pathlib import Path import pytest -from autoskillit.core import RetryReason, WriteBehaviorSpec, extract_skill_name +from autoskillit.core import ( + DelegatedWriteTarget, + RetryReason, + WriteBehaviorSpec, + extract_skill_name, +) from autoskillit.execution.backends.claude import ClaudeCodeBackend from autoskillit.execution.headless import _build_skill_result from tests.conftest import _make_result +from tests.execution.conftest import ( + delegated_child_prompt, + delegated_child_tool_result, + delegated_child_tool_use, + delegated_result_record, + delegated_root_agent_call, + delegated_root_agent_result, + delegated_root_text, + delegated_stream, + delegated_task_notification, + delegated_task_started, +) pytestmark = [pytest.mark.layer("execution"), pytest.mark.small] @@ -979,21 +997,108 @@ def _always_write_skills() -> list[str]: @pytest.mark.parametrize("skill_name", _always_write_skills.__func__()) # type: ignore[attr-defined] def test_always_write_skill_with_edits_passes(self, skill_name: str) -> None: - stdout = _ndjson_with_tool_uses(["Edit", "Write"]) + from autoskillit.recipe.contracts import get_skill_contract, load_bundled_manifest + + manifest = load_bundled_manifest() + contract = get_skill_contract(skill_name, manifest) + assert contract is not None + assert contract.write_behavior is not None + stdout = _ndjson_with_tool_uses(["Edit"]) sr = _build_skill_result( _make_result(returncode=0, stdout=stdout), skill_command=f"/autoskillit:{skill_name} arg", - write_behavior=WriteBehaviorSpec(mode="always"), + write_behavior=WriteBehaviorSpec( + mode=contract.write_behavior, + expected_when=tuple(contract.write_expected_when), + external_effect=contract.external_effect, + ), backend=ClaudeCodeBackend(), ) assert sr.evidence.has_evidence is True, ( f"Skill {skill_name!r} (write_behavior=always) must detect write evidence" ) + assert sr.evidence.has_implementation_evidence is True assert sr.subtype != "zero_writes", ( f"Skill {skill_name!r} must not trigger zero_writes gate when writes are present" ) +class TestContractDrivenDelegatedWriteSkills: + """Every manifest-declared delegated-write input admits a changed owned edit.""" + + @staticmethod + def _delegated_write_skills() -> list[str]: + from autoskillit.recipe.contracts import load_bundled_manifest + + manifest = load_bundled_manifest() + skills = manifest.get("skills", {}) + if not isinstance(skills, dict): + return [] + names = [ + name + for name, contract in skills.items() + if ( + contract.get("delegated_write_input") + if isinstance(contract, dict) + else getattr(contract, "delegated_write_input", None) + ) + ] + assert names, "load_bundled_manifest() returned no delegated-write contracts" + return names + + @pytest.mark.parametrize( + "skill_name", + _delegated_write_skills.__func__(), # type: ignore[attr-defined] + ) + def test_manifest_delegated_write_contract_admits_child_edit( + self, tmp_path: Path, skill_name: str + ) -> None: + from autoskillit.recipe.contracts import get_skill_contract, load_bundled_manifest + + plan_dir = tmp_path / ".autoskillit" / "temp" / "make-plan" + plan_dir.mkdir(parents=True) + plan = plan_dir / "plan.md" + plan.write_text("Plan draft\n") + target = DelegatedWriteTarget.bind(os.path.realpath(plan)) + plan.write_text("Dry-walkthrough verified = TRUE\n" + plan.read_text()) + contract = get_skill_contract(skill_name, load_bundled_manifest()) + assert contract is not None + assert contract.delegated_write_input + assert contract.write_behavior is not None + stdout = delegated_stream( + [ + delegated_root_agent_call(), + delegated_task_started(), + delegated_child_prompt(), + delegated_child_tool_use(file_path=str(plan)), + delegated_child_tool_result(), + delegated_task_notification(), + delegated_root_agent_result(), + delegated_root_text(), + delegated_result_record(), + ] + ) + + sr = _build_skill_result( + _make_result(returncode=0, stdout=stdout), + skill_command=f"/autoskillit:{skill_name} {plan}", + cwd=str(tmp_path), + write_watch_dirs=(plan_dir,), + write_behavior=WriteBehaviorSpec( + mode=contract.write_behavior, + expected_when=tuple(contract.write_expected_when), + external_effect=contract.external_effect, + delegated_target=target, + ), + skill_contract=contract, + backend=ClaudeCodeBackend(), + ) + + assert sr.subtype != "zero_writes" + assert sr.evidence.write_call_count == 0 + assert sr.evidence.delegated_write_count == 1 + + class TestStdoutWriteHeuristicStrengthening: """Heuristic guards: truncated-line fallback must not fire on false positives.""" @@ -1106,6 +1211,33 @@ def test_temp_only_writes_not_implementation_evidence_temp_fallback( ) assert sr.evidence.has_implementation_evidence is False + def test_worktree_write_without_file_path_is_not_implementation_evidence( + self, tmp_path: Path + ) -> None: + wt = self._worktree_cwd(tmp_path) + stdout = _ndjson_with_tool_uses(["Write"]) + sr = _build_skill_result( + _make_result(returncode=0, stdout=stdout), + skill_command="/autoskillit:implement-worktree-no-merge /some/path.md", + cwd=str(wt), + write_watch_dirs=[wt], + write_behavior=WriteBehaviorSpec(mode="always"), + backend=ClaudeCodeBackend(), + ) + assert sr.evidence.write_call_count == 0 + + def test_non_worktree_write_without_file_path_is_counted(self, tmp_path: Path) -> None: + wt = self._worktree_cwd(tmp_path) + stdout = _ndjson_with_tool_uses(["Write"]) + sr = _build_skill_result( + _make_result(returncode=0, stdout=stdout), + skill_command="/autoskillit:investigate regression", + cwd=str(wt), + write_behavior=WriteBehaviorSpec(mode="always"), + backend=ClaudeCodeBackend(), + ) + assert sr.evidence.write_call_count == 1 + def test_mixed_temp_and_tracked_writes_is_implementation_evidence( self, tmp_path: Path ) -> None: diff --git a/tests/recipe/test_delegated_write_contracts.py b/tests/recipe/test_delegated_write_contracts.py new file mode 100644 index 0000000000..b8d12e52e7 --- /dev/null +++ b/tests/recipe/test_delegated_write_contracts.py @@ -0,0 +1,47 @@ +"""Manifest declarations that authorize a child plan edit.""" + +from __future__ import annotations + +import pytest + +from autoskillit.recipe.contracts import get_skill_contract, load_bundled_manifest + +pytestmark = [pytest.mark.layer("recipe"), pytest.mark.small] + + +def test_dry_walkthrough_declares_its_delegated_write_input() -> None: + contract = get_skill_contract("dry-walkthrough", load_bundled_manifest()) + + assert contract is not None + assert contract.delegated_write_input == "plan_path" + + +def test_every_bundled_skill_contract_parses() -> None: + manifest = load_bundled_manifest() + + for skill_name in manifest["skills"]: + assert get_skill_contract(skill_name, manifest) is not None, skill_name + + +@pytest.mark.parametrize( + ("inputs", "delegated_input", "write_behavior"), + [ + ([{"name": "plan_path", "type": "file_path"}], "unknown_path", "always"), + ([{"name": "plan_path", "type": "string"}], "plan_path", "always"), + ([{"name": "plan_path", "type": "file_path"}], "plan_path", None), + ([{"name": "plan_path", "type": "file_path"}], "plan_path", ["always"]), + ([{"name": "plan_path", "type": "file_path"}], 1, "always"), + ], +) +def test_invalid_delegated_write_declarations_name_the_skill( + inputs: list[dict[str, object]], + delegated_input: object, + write_behavior: object, +) -> None: + skill_data: dict[str, object] = {"inputs": inputs, "outputs": []} + skill_data["delegated_write_input"] = delegated_input + if write_behavior is not None: + skill_data["write_behavior"] = write_behavior + + with pytest.raises(ValueError, match="example-skill"): + get_skill_contract("example-skill", {"skills": {"example-skill": skill_data}}) diff --git a/tests/server/test_audit_cycle_delivery_integration.py b/tests/server/test_audit_cycle_delivery_integration.py index cd6e8e9a37..6271a72555 100644 --- a/tests/server/test_audit_cycle_delivery_integration.py +++ b/tests/server/test_audit_cycle_delivery_integration.py @@ -109,12 +109,13 @@ def test_audit_response_exposes_reserved_semantic_path(tmp_path: Path) -> None: def test_prompt_contract_mode_is_bound_before_child_dispatch() -> None: - standalone = _recipe_execution.build_standalone_child_prompt( + standalone, canonical_inputs = _recipe_execution.build_standalone_child_invocation( "/autoskillit:audit-impl", "/tmp", None, audit_output_mode=_recipe_execution.AuditOutputMode.STANDALONE, ) + assert canonical_inputs == () standalone_payload = json.loads(standalone.split("AUTOSKILLIT_BOUND_INVOCATION_V1\n", 1)[1]) assert standalone_payload["audit_output_mode"] == "standalone" assert "audit_semantic_submission" not in standalone_payload diff --git a/tests/server/test_run_skill_delegated_write_binding.py b/tests/server/test_run_skill_delegated_write_binding.py new file mode 100644 index 0000000000..62a2090141 --- /dev/null +++ b/tests/server/test_run_skill_delegated_write_binding.py @@ -0,0 +1,209 @@ +"""Binding a declared child write to the canonical invocation input.""" + +from __future__ import annotations + +import hashlib +from pathlib import Path +from types import SimpleNamespace + +import pytest +import structlog.testing + +from autoskillit.core import WriteBehaviorSpec +from autoskillit.recipe.contracts import get_skill_contract, load_bundled_manifest +from autoskillit.server.tools._execution_helpers import ( + deserialize_skill_contract, + resolve_skill_dispatch_metadata, + serialize_skill_contract, +) +from autoskillit.server.tools.tools_execution._run_skill_session import ( + _bind_delegated_target, +) + +pytestmark = [pytest.mark.layer("server"), pytest.mark.small] + + +def _binding_state(cwd: Path, bound_inputs: tuple[tuple[str, object], ...]) -> SimpleNamespace: + return SimpleNamespace( + skill_command="/autoskillit:dry-walkthrough", + _skill_contract=SimpleNamespace(delegated_write_input="plan_path"), + _stored_contract=None, + write_spec=WriteBehaviorSpec(mode="always"), + _bound_recipe_inputs=bound_inputs, + cwd=str(cwd), + ) + + +@pytest.mark.parametrize("path_form", ["relative", "absolute", "ancestor_symlink"]) +def test_binder_captures_in_scope_plan_digest(tmp_path: Path, path_form: str) -> None: + root = tmp_path / "root" + nested = root / "nested" + nested.mkdir(parents=True) + plan = nested / "plan.md" + plan.write_text("plan before dispatch", encoding="utf-8") + (root / "alias").symlink_to(nested, target_is_directory=True) + input_path = { + "relative": "nested/plan.md", + "absolute": str(plan), + "ancestor_symlink": "alias/plan.md", + }[path_form] + + state = _binding_state(root, (("plan_path", input_path),)) + _bind_delegated_target(state) + + target = state.write_spec.delegated_target + assert target.path == str(plan.resolve()) + assert target.baseline_digest == hashlib.sha256(plan.read_bytes()).hexdigest() + + +@pytest.mark.parametrize( + "case", + ["absent", "empty", "non_string", "outside", "null_byte", "final_symlink"], +) +def test_invalid_target_stays_unbound_and_logs_reason(tmp_path: Path, case: str) -> None: + root = tmp_path / "root" + root.mkdir() + plan = root / "plan.md" + plan.write_text("plan", encoding="utf-8") + outside = tmp_path / "outside.md" + outside.write_text("outside", encoding="utf-8") + (root / "link.md").symlink_to(plan) + inputs: dict[str, tuple[tuple[str, object], ...]] = { + "absent": (), + "empty": (("plan_path", ""),), + "non_string": (("plan_path", 42),), + "outside": (("plan_path", str(outside)),), + "null_byte": (("plan_path", "bad\x00path"),), + "final_symlink": (("plan_path", "link.md"),), + } + state = _binding_state(root, inputs[case]) + + with structlog.testing.capture_logs() as events: + _bind_delegated_target(state) + + assert state.write_spec.delegated_target is None + assert any(event.get("event") == "delegated_write_target_unbound" for event in events) + + +def test_unreadable_baseline_stays_unbound_and_logs_reason(tmp_path: Path, monkeypatch) -> None: + plan = tmp_path / "plan.md" + plan.write_text("plan") + state = _binding_state(tmp_path, (("plan_path", "plan.md"),)) + + def unreadable(path: Path) -> bytes: + raise PermissionError("baseline unavailable") + + monkeypatch.setattr(Path, "read_bytes", unreadable) + with structlog.testing.capture_logs() as events: + _bind_delegated_target(state) + + assert state.write_spec.delegated_target is None + assert any( + event.get("event") == "delegated_write_target_unbound" + and "baseline unavailable" in event.get("reason", "") + for event in events + ) + + +def test_contract_without_declaration_leaves_write_spec_unchanged(tmp_path: Path) -> None: + state = _binding_state(tmp_path, (("plan_path", "plan.md"),)) + state._skill_contract = SimpleNamespace(delegated_write_input=None) + original_spec = state.write_spec + + _bind_delegated_target(state) + + assert state.write_spec is original_spec + + +def _restored_state( + cwd: Path, + skill_contract: object, + write_spec: WriteBehaviorSpec, +) -> SimpleNamespace: + stored_contract = SimpleNamespace( + expected_output_patterns=[], + write_behavior=write_spec, + skill_contract_json=serialize_skill_contract(skill_contract), + ) + _, restored_spec, restored_contract = resolve_skill_dispatch_metadata( + None, "/autoskillit:dry-walkthrough", stored_contract + ) + return SimpleNamespace( + _skill_contract=restored_contract, + _stored_contract=stored_contract, + write_spec=restored_spec, + _bound_recipe_inputs=(("plan_path", "plan.md"),), + cwd=str(cwd), + ) + + +def test_resume_keeps_original_target_baseline_without_rebinding(tmp_path: Path) -> None: + from autoskillit.core import DelegatedWriteTarget + + plan = tmp_path / "plan.md" + plan.write_text("original", encoding="utf-8") + original_target = DelegatedWriteTarget.bind(str(plan)) + plan.write_text("changed after binding", encoding="utf-8") + contract = get_skill_contract("dry-walkthrough", load_bundled_manifest()) + assert contract is not None + state = _restored_state( + tmp_path, + contract, + WriteBehaviorSpec(mode="always", delegated_target=original_target), + ) + + _bind_delegated_target(state) + + restored_target = state.write_spec.delegated_target + assert restored_target.baseline_digest == original_target.baseline_digest + assert restored_target.changed() + + +def test_resume_without_target_stays_unbound_without_warning(tmp_path: Path) -> None: + contract = get_skill_contract("dry-walkthrough", load_bundled_manifest()) + assert contract is not None + state = _restored_state(tmp_path, contract, WriteBehaviorSpec(mode="always")) + + with structlog.testing.capture_logs() as events: + _bind_delegated_target(state) + + assert state.write_spec.delegated_target is None + assert not any(event.get("event") == "delegated_write_target_unbound" for event in events) + + +@pytest.mark.anyio +@pytest.mark.parametrize("structured", [True, False]) +async def test_run_skill_binds_structured_and_positional_plan_inputs( + tool_ctx_kitchen_open, + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + structured: bool, +) -> None: + from autoskillit.server.tools.tools_execution import run_skill + from tests.fakes import InMemoryHeadlessExecutor + + plan = tmp_path / "plan.md" + plan.write_text("plan", encoding="utf-8") + executor = InMemoryHeadlessExecutor() + monkeypatch.setattr(tool_ctx_kitchen_open, "executor", executor) + monkeypatch.setattr("autoskillit.server._ctx", tool_ctx_kitchen_open) + skill_command = ( + "/autoskillit:dry-walkthrough" if structured else f"/autoskillit:dry-walkthrough {plan}" + ) + skill_inputs = {"plan_path": str(plan)} if structured else None + + await run_skill(skill_command, str(tmp_path), skill_inputs=skill_inputs) + + assert len(executor.calls) == 1 + target = executor.calls[0].write_behavior.delegated_target + assert target.path == str(plan.resolve()) + + +def test_serialized_skill_contract_preserves_delegated_write_input() -> None: + contract = get_skill_contract("dry-walkthrough", load_bundled_manifest()) + assert contract is not None + + restored = deserialize_skill_contract(serialize_skill_contract(contract)) + + assert restored is not None + assert restored.delegated_write_input == "plan_path" diff --git a/tests/server/test_tools_execution_write_prefix.py b/tests/server/test_tools_execution_write_prefix.py index fd8f2e72e0..661a85868a 100644 --- a/tests/server/test_tools_execution_write_prefix.py +++ b/tests/server/test_tools_execution_write_prefix.py @@ -2,6 +2,7 @@ from __future__ import annotations +import json from pathlib import Path import pytest @@ -242,7 +243,10 @@ async def test_pass_when_scope_covers_cwd(tool_ctx_kitchen_open, monkeypatch, tm call = executor.calls[0] assert call.capability_contract is not None assert not hasattr(call.capability_contract, "resolved_command") - assert call.skill_command == f"/implement-worktree-no-merge {plan_path}" + invocation_prefix = "\n\nAUTOSKILLIT_BOUND_INVOCATION_V1\n" + command, payload = call.skill_command.split(invocation_prefix, 1) + assert command == f"/implement-worktree-no-merge {plan_path}" + assert json.loads(payload)["skill_inputs"] == [{"name": "plan_path", "value": str(plan_path)}] assert call.capability_contract.cwd == str(worktree.resolve()) assert call.cwd == str(worktree.resolve()) diff --git a/tests/test_test_filter_core_cascade.py b/tests/test_test_filter_core_cascade.py index acb72730ce..e1c50df7f9 100644 --- a/tests/test_test_filter_core_cascade.py +++ b/tests/test_test_filter_core_cascade.py @@ -152,6 +152,7 @@ def test_all_entries_present(self) -> None: "_type_checkpoint", "_type_results", "_type_results_records", + "_type_write_evidence", "_type_execution_identity", "_type_exploration", "_type_results_execution", @@ -783,6 +784,7 @@ def test_type_checkpoint_narrow_routing(self, tmp_path: Path) -> None: [ "src/autoskillit/core/types/results/_type_results.py", "src/autoskillit/core/types/results/_type_results_records.py", + "src/autoskillit/core/types/results/_type_write_evidence.py", ], ) def test_type_results_narrow_routing(self, tmp_path: Path, source_path: str) -> None: