fix: credit owned child writes to fresh declared outputs

src/autoskillit/core/__init__.pyi

@@ -964,6 +964,7 @@ 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 DelegatedOutputScope as DelegatedOutputScope
 from .types import DelegatedWriteObservation as DelegatedWriteObservation
 from .types import DelegatedWriteTarget as DelegatedWriteTarget
 from .types import DeliveryOccurrenceId as DeliveryOccurrenceId

src/autoskillit/core/types/results/_type_results.py

@@ -65,7 +65,7 @@
     require_aware_timestamp,
 )
 from ._type_token import TurnTokenEntry
-from ._type_write_evidence import DelegatedWriteTarget, WriteEvidence
+from ._type_write_evidence import DelegatedOutputScope, DelegatedWriteTarget, WriteEvidence
 
 if TYPE_CHECKING:
     from ..launch._type_launch import ResolvedLaunchContract
@@ -248,12 +248,14 @@ class WriteBehaviorSpec:
         ``none``, ``serialized-idempotent``, or
         ``serialized-unknown-completion`` for effects outside the workspace.
     delegated_target: Canonical in-scope target for delegated writes.
+    delegated_output_scope: Attempt-local scope for fresh declared-output writes.
     """
 
     mode: str | None = None
     expected_when: tuple[str, ...] = ()
     external_effect: str = "none"
     delegated_target: DelegatedWriteTarget | None = None
+    delegated_output_scope: DelegatedOutputScope | None = None
 
     def __post_init__(self) -> None:
         if self.external_effect not in VALID_EXTERNAL_EFFECTS:

src/autoskillit/core/types/results/_type_write_evidence.py

@@ -7,7 +7,12 @@
 from pathlib import Path
 from stat import S_ISREG
 
-__all__ = ["DelegatedWriteObservation", "DelegatedWriteTarget", "WriteEvidence"]
+__all__ = [
+    "DelegatedOutputScope",
+    "DelegatedWriteObservation",
+    "DelegatedWriteTarget",
+    "WriteEvidence",
+]
 
 
 @dataclass(frozen=True, slots=True)
@@ -46,6 +51,24 @@ def changed(self) -> bool:
         return current_digest is not None and current_digest != self.baseline_digest
 
 
+@dataclass(frozen=True, slots=True)
+class DelegatedOutputScope:
+    """Canonical output root and regular files observed during a completed scan.
+
+    The baseline is not an atomic snapshot or an attestation of artifact bytes.
+    """
+
+    root: str
+    preexisting: frozenset[str]
+
+    def target_for(self, path: str) -> DelegatedWriteTarget | None:
+        if path == self.root or not Path(path).is_relative_to(self.root):
+            return None
+        if path in self.preexisting:
+            return None
+        return DelegatedWriteTarget(path=path, baseline_digest=None)
+
+
 @dataclass(frozen=True, slots=True)
 class DelegatedWriteObservation:
     """A parsed child write-tool observation."""
@@ -60,13 +83,17 @@ class DelegatedWriteObservation:
 
 @dataclass(frozen=True, slots=True)
 class WriteEvidence:
-    """Bundled write evidence signals — either explicitly constructed or absent."""
+    """Bundled write evidence signals — either explicitly constructed or absent.
+
+    Owned child paths await declared-output admission and are never evidence alone.
+    """
 
     write_call_count: int
     fs_writes_detected: bool
     git_writes_detected: bool
     file_changes_count: int = 0
     delegated_write_count: int = 0
+    delegated_write_paths: tuple[str, ...] = ()
 
     @classmethod
     def none_observed(cls) -> WriteEvidence:

src/autoskillit/execution/headless/_headless_adjudication.py

@@ -59,6 +59,7 @@
 from autoskillit.execution.session._session_outcome import _compute_success
 from autoskillit.execution.session._write_evidence import (
     WriteEvidenceScope,
+    _admit_declared_output_write,
     _compute_write_evidence,
 )
 
@@ -307,32 +308,43 @@ def _apply_post_session_adjudication(
     if not sr.success:
         return sr
 
-    if not evidence.has_implementation_evidence and write_behavior is not None:
-        write_expected = False
+    write_expected = False
+    if write_behavior is not None:
         if write_behavior.mode == "always":
             write_expected = True
         elif write_behavior.mode == "conditional" and write_behavior.expected_when:
             write_expected = _check_expected_patterns(
                 sr.result,
                 write_behavior.expected_when,
             )
-        if write_expected:
-            detail = "expected implementation writes were not observed"
-            return dataclasses.replace(
-                sr,
-                success=False,
+    if write_expected:
+        admitted = _admit_declared_output_write(
+            evidence,
+            fields,
+            skill_contract.delegated_write_output if skill_contract else None,
+            write_behavior.delegated_output_scope if write_behavior else None,
+            cwd,
+        )
+        if admitted is not evidence:
+            evidence = admitted
+            sr = dataclasses.replace(sr, evidence=admitted)
+    if write_expected and not evidence.has_implementation_evidence:
+        detail = "expected implementation writes were not observed"
+        return dataclasses.replace(
+            sr,
+            success=False,
+            subtype="zero_writes",
+            needs_retry=True,
+            retry_reason=RetryReason.ZERO_WRITES,
+            result=detail,
+            adjudication_verdict=AdjudicationVerdict(
+                reason_kind=RetryReason.ZERO_WRITES,
                 subtype="zero_writes",
-                needs_retry=True,
-                retry_reason=RetryReason.ZERO_WRITES,
-                result=detail,
-                adjudication_verdict=AdjudicationVerdict(
-                    reason_kind=RetryReason.ZERO_WRITES,
-                    subtype="zero_writes",
-                    detail=detail,
-                    outcome_fields=sr.outcome_fields,
-                    defects=(),
-                ),
-            )
+                detail=detail,
+                outcome_fields=sr.outcome_fields,
+                defects=(),
+            ),
+        )
 
     if skill_contract is None:
         return sr

src/autoskillit/execution/headless/_headless_result.py

@@ -299,6 +299,7 @@ def _build_skill_result(
         cwd=cwd,
         skill_command=skill_command,
         delegated_target=getattr(write_behavior, "delegated_target", None),
+        delegated_output_scope=getattr(write_behavior, "delegated_output_scope", None),
     )
     context = _SkillResultContext(
         backend=backend,

src/autoskillit/execution/session/_write_evidence.py

@@ -4,12 +4,13 @@
 
 import os
 from collections.abc import Collection
-from dataclasses import dataclass
+from dataclasses import dataclass, replace
 from pathlib import Path
 from typing import Any
 
 from autoskillit.core import (
     WORKTREE_SKILLS,
+    DelegatedOutputScope,
     DelegatedWriteObservation,
     DelegatedWriteTarget,
     WriteEvidence,
@@ -31,6 +32,7 @@ class WriteEvidenceScope:
     cwd: str
     skill_command: str
     delegated_target: DelegatedWriteTarget | None
+    delegated_output_scope: DelegatedOutputScope | None
 
 
 def _is_worktree_dispatch(scope: WriteEvidenceScope) -> bool:
@@ -83,15 +85,23 @@ def _count_root_write_calls(session: ClaudeSessionResult, scope: WriteEvidenceSc
     return _count_worktree_write_calls(scope, eligible_calls)
 
 
-def _is_admitted_delegated_write(
+def _canonical_path(cwd: str, file_path: str) -> str | None:
+    try:
+        return os.path.realpath(Path(cwd) / file_path)
+    except (OSError, ValueError):
+        return None
+
+
+def _qualifying_delegated_write_path(
     observation: DelegatedWriteObservation,
-    target: DelegatedWriteTarget,
     scope: WriteEvidenceScope,
-) -> bool:
+) -> str | None:
     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)
+        return None
+    resolved = _canonical_path(scope.cwd, observation.file_path)
+    if resolved is None or not _is_implementation_path(scope, resolved):
+        return None
+    return resolved
 
 
 def _admitted_delegated_write_count(
@@ -120,12 +130,12 @@ def _admitted_delegated_write_count(
         admitted = sum(
             1
             for observation in observations
-            if _is_admitted_delegated_write(observation, target, scope)
+            if _qualifying_delegated_write_path(observation, scope) == target.path
         )
     else:
         admitted = 0
 
-    if admitted == 0:
+    if admitted == 0 and scope.delegated_output_scope is None:
         logger.warning(
             "delegated_writes_not_admitted",
             observed=len(observations),
@@ -135,6 +145,51 @@ def _admitted_delegated_write_count(
     return admitted
 
 
+def _owned_write_paths(session: ClaudeSessionResult, scope: WriteEvidenceScope) -> tuple[str, ...]:
+    if scope.delegated_output_scope is None:
+        return ()
+    return tuple(
+        path
+        for observation in session.delegated_writes
+        if (path := _qualifying_delegated_write_path(observation, scope)) is not None
+    )
+
+
+def _admit_declared_output_write(
+    evidence: WriteEvidence,
+    fields: dict[str, str | int],
+    output_name: str | None,
+    output_scope: DelegatedOutputScope | None,
+    cwd: str,
+) -> WriteEvidence:
+    if not output_name or output_scope is None or not evidence.delegated_write_paths:
+        return evidence
+    value = fields.get(output_name)
+    path = _canonical_path(cwd, value) if isinstance(value, str) and value else None
+    target = output_scope.target_for(path) if path else None
+    admitted = 0
+    try:
+        if target is not None and target.changed():
+            admitted = evidence.delegated_write_paths.count(target.path)
+    except OSError as exc:
+        logger.debug(
+            "delegated_output_freshness_check_failed",
+            output=output_name,
+            error_type=type(exc).__name__,
+        )
+    if admitted == 0:
+        if not evidence.has_implementation_evidence:
+            logger.warning(
+                "delegated_output_write_not_admitted",
+                output=output_name,
+                observed=len(evidence.delegated_write_paths),
+                declared=path is not None,
+                in_scope=target is not None,
+            )
+        return evidence
+    return replace(evidence, delegated_write_count=evidence.delegated_write_count + admitted)
+
+
 def _compute_write_evidence(
     session: ClaudeSessionResult,
     scope: WriteEvidenceScope,
@@ -165,4 +220,5 @@ def _compute_write_evidence(
         git_writes_detected=scope.git_writes_detected,
         file_changes_count=file_changes_count,
         delegated_write_count=delegated_write_count,
+        delegated_write_paths=_owned_write_paths(session, scope),
     )

src/autoskillit/recipe/contracts/_contracts_manifest.py

@@ -5,7 +5,7 @@
 import dataclasses
 import hashlib
 import json
-from collections.abc import Callable, Mapping
+from collections.abc import Callable, Mapping, Sequence
 from pathlib import Path
 from typing import Any, Literal, assert_never, cast
 
@@ -257,20 +257,22 @@ def _as_input_preflight(value: object) -> tuple[str, ...]:
     return tuple(value)
 
 
-def _parse_delegated_write_input(
+def _parse_delegated_write_name(
     skill_name: str,
     skill_data: dict[str, Any],
-    inputs: tuple[SkillInput, ...],
+    key: Literal["delegated_write_input", "delegated_write_output"],
+    declared: Sequence[SkillInput | SkillOutput],
     write_behavior: object,
 ) -> str | None:
-    if "delegated_write_input" not in skill_data:
+    if key not in skill_data:
         return None
-    value = skill_data["delegated_write_input"]
-    error_prefix = f"delegated_write_input for skill '{skill_name}'"
+    value = skill_data[key]
+    error_prefix = f"{key} 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 not any(item.name == value and item.type == "file_path" for item in declared):
+        kind = "input" if key == "delegated_write_input" else "output"
+        raise ValueError(f"{error_prefix} must name a declared file_path {kind}")
     if write_behavior not in ("always", "conditional"):
         raise ValueError(f"{error_prefix} requires write_behavior 'always' or 'conditional'")
     return value
@@ -295,8 +297,11 @@ 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
+    delegated_write_input = _parse_delegated_write_name(
+        skill_name, skill_data, "delegated_write_input", inputs, write_behavior
+    )
+    delegated_write_output = _parse_delegated_write_name(
+        skill_name, skill_data, "delegated_write_output", outputs, write_behavior
     )
     external_effect = skill_data.get("external_effect", "none")
     if not isinstance(external_effect, str):
@@ -336,6 +341,7 @@ def get_skill_contract(skill_name: str, manifest: dict[str, Any]) -> SkillContra
         pattern_examples=examples,
         write_behavior=write_behavior,
         delegated_write_input=delegated_write_input,
+        delegated_write_output=delegated_write_output,
         write_expected_when=write_expected_when,
         external_effect=external_effect,
         read_only=read_only,

src/autoskillit/recipe/contracts/_contracts_types.py

@@ -134,6 +134,7 @@ class SkillContract:
     )
     audit_output_mode: AuditOutputMode | None = None
     delegated_write_input: str | None = None
+    delegated_write_output: str | None = None
 
     def __post_init__(self) -> None:
         if self.external_effect not in VALID_EXTERNAL_EFFECTS:

src/autoskillit/recipe/skill_contracts.yaml

@@ -713,6 +713,7 @@ skills:
     - "https://github.com/owner/repo/pull/42\n%%ORDER_UP%%"
 
   prepare-pr:
+    delegated_write_output: prep_path
     inputs:
     - name: plan_paths
       type: file_path_list

src/autoskillit/server/tools/_execution_helpers/_skill_contract.py

@@ -258,6 +258,7 @@ def deserialize_skill_contract(payload: str) -> SkillContract | None:
             pattern_examples=list(data.get("pattern_examples", [])),
             write_behavior=data.get("write_behavior"),
             delegated_write_input=data.get("delegated_write_input"),
+            delegated_write_output=data.get("delegated_write_output"),
             write_expected_when=list(data.get("write_expected_when", [])),
             external_effect=str(data.get("external_effect", "none")),
             read_only=read_only,