Fix delegated write evidence for owned child sessions

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

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
+    )

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."""

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
+        )

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,

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 []