diff --git a/src/autoskillit/core/__init__.pyi b/src/autoskillit/core/__init__.pyi index 022d54576c..69febdc098 100644 --- a/src/autoskillit/core/__init__.pyi +++ b/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 diff --git a/src/autoskillit/core/types/results/_type_results.py b/src/autoskillit/core/types/results/_type_results.py index 39f9aad68a..a351e63198 100644 --- a/src/autoskillit/core/types/results/_type_results.py +++ b/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: diff --git a/src/autoskillit/core/types/results/_type_write_evidence.py b/src/autoskillit/core/types/results/_type_write_evidence.py index 7ed447ac10..8464b41bec 100644 --- a/src/autoskillit/core/types/results/_type_write_evidence.py +++ b/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: diff --git a/src/autoskillit/execution/headless/_headless_adjudication.py b/src/autoskillit/execution/headless/_headless_adjudication.py index c12a2efbca..95a6964fbc 100644 --- a/src/autoskillit/execution/headless/_headless_adjudication.py +++ b/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,8 +308,8 @@ 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: @@ -316,23 +317,34 @@ def _apply_post_session_adjudication( 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 diff --git a/src/autoskillit/execution/headless/_headless_result.py b/src/autoskillit/execution/headless/_headless_result.py index 4418771ec4..00944502b5 100644 --- a/src/autoskillit/execution/headless/_headless_result.py +++ b/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, diff --git a/src/autoskillit/execution/session/_write_evidence.py b/src/autoskillit/execution/session/_write_evidence.py index 68c02fec0b..d043caf7c6 100644 --- a/src/autoskillit/execution/session/_write_evidence.py +++ b/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), ) diff --git a/src/autoskillit/recipe/contracts/_contracts_manifest.py b/src/autoskillit/recipe/contracts/_contracts_manifest.py index 365b3886ee..a984693817 100644 --- a/src/autoskillit/recipe/contracts/_contracts_manifest.py +++ b/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, diff --git a/src/autoskillit/recipe/contracts/_contracts_types.py b/src/autoskillit/recipe/contracts/_contracts_types.py index d8ab8f66c3..dc6d43fe3b 100644 --- a/src/autoskillit/recipe/contracts/_contracts_types.py +++ b/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: diff --git a/src/autoskillit/recipe/skill_contracts.yaml b/src/autoskillit/recipe/skill_contracts.yaml index 156ac61ec0..dcde2ab309 100644 --- a/src/autoskillit/recipe/skill_contracts.yaml +++ b/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 diff --git a/src/autoskillit/server/tools/_execution_helpers/_skill_contract.py b/src/autoskillit/server/tools/_execution_helpers/_skill_contract.py index 22bbe6c420..d4c642e6eb 100644 --- a/src/autoskillit/server/tools/_execution_helpers/_skill_contract.py +++ b/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, 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 9211de5a3c..166b0df951 100644 --- a/src/autoskillit/server/tools/tools_execution/_run_skill_session.py +++ b/src/autoskillit/server/tools/tools_execution/_run_skill_session.py @@ -21,6 +21,7 @@ SKILL_COMMAND_DISPLAY_MAX, WORKTREE_SKILLS, ContainmentError, + DelegatedOutputScope, DelegatedWriteTarget, SkillContractError, SkillExecutionRole, @@ -29,6 +30,7 @@ extract_skill_name, get_logger, resolve_contained_path, + strict_walk, ) from autoskillit.core import current_order_id as _current_order_id from autoskillit.core import current_step_name as _current_step_name @@ -371,7 +373,7 @@ async def _prepare_owned_dispatch_session( if (terminal := _extend_closure_write_scope(state)) is not None: return terminal - _bind_delegated_target(state) + _bind_delegated_writes(state) _prepare_owned_session_contract(state) if (terminal := _prepare_owned_write_scope(state)) is not None: @@ -531,10 +533,14 @@ 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.""" +def _bind_delegated_writes(state: _RunSkillDispatchState) -> None: + """Bind declared child-write targets for this invocation.""" + if state.write_spec is None: + return + if getattr(state._skill_contract, "delegated_write_output", None): + _bind_delegated_output_scope(state) 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: + if state._stored_contract is not None or not input_name: return value = dict(state._bound_recipe_inputs).get(input_name) @@ -565,6 +571,43 @@ def _bind_delegated_target(state: _RunSkillDispatchState) -> None: ) +def _bind_delegated_output_scope(state: _RunSkillDispatchState) -> None: + assert state.write_spec is not None + if state.output_dir: + reason = "explicit_output_dir" + elif state._default_temp is None: + reason = "no_output_root" + else: + try: + root = os.path.realpath(state._default_temp) + if not Path(root).is_relative_to(os.path.realpath(state.cwd)): + reason = "output_root_outside_cwd" + else: + try: + os.lstat(root) + except FileNotFoundError: + preexisting: frozenset[str] = frozenset() + else: + preexisting = frozenset( + os.path.join(root, entry.relative_path) + for entry in strict_walk(Path(root)) + if entry.kind == "f" + ) + state.write_spec = replace( + state.write_spec, + delegated_output_scope=DelegatedOutputScope(root, preexisting), + ) + return + except (OSError, ValueError, NotImplementedError) as exc: + reason = str(exc) + logger.warning( + "delegated_write_target_unbound", + skill=state.skill_command, + output=getattr(state._skill_contract, "delegated_write_output", None), + 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 0b6d74d406..864b034f52 100644 --- a/tests/_test_filter.py +++ b/tests/_test_filter.py @@ -1336,6 +1336,7 @@ class SourceMapRejection(enum.StrEnum): "hooks/test_recipe_contract_freshness.py", "migration", # Server file-level entries importing autoskillit.recipe: + "server/test_run_skill_delegated_output_write.py", "server/test_run_skill_execution_tuning_fallbacks.py", "server/test_run_skill_delegated_write_binding.py", "server/test_bundled_dispatch_write_scope_admission.py", diff --git a/tests/contracts/test_skill_result_persistence_ledger.py b/tests/contracts/test_skill_result_persistence_ledger.py index 8badcf894b..fe05eafc39 100644 --- a/tests/contracts/test_skill_result_persistence_ledger.py +++ b/tests/contracts/test_skill_result_persistence_ledger.py @@ -57,6 +57,7 @@ ("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.delegated_write_paths", "delegated_write_paths", "live-only:ephemeral-path"), ("evidence.file_changes_count", "file_changes_count", "persisted"), ("evidence.fs_writes_detected", "fs_writes_detected", "persisted"), ("evidence.git_writes_detected", "git_writes_detected", "persisted"), diff --git a/tests/core/types/test_type_results.py b/tests/core/types/test_type_results.py index 4db12add62..cfd94616ba 100644 --- a/tests/core/types/test_type_results.py +++ b/tests/core/types/test_type_results.py @@ -97,6 +97,24 @@ def test_has_implementation_evidence_true_with_file_changes(self) -> None: ) assert ev.has_implementation_evidence is True + def test_delegated_write_paths_alone_are_not_evidence(self) -> None: + defaults = WriteEvidence( + write_call_count=0, + fs_writes_detected=False, + git_writes_detected=False, + ) + paths_only = WriteEvidence( + write_call_count=0, + fs_writes_detected=False, + git_writes_detected=False, + delegated_write_paths=("/workspace/plan.md",), + ) + + assert defaults.delegated_write_paths == () + assert paths_only.delegated_write_paths == ("/workspace/plan.md",) + assert paths_only.has_implementation_evidence is False + assert paths_only.has_evidence is False + class TestDelegatedWriteTarget: @pytest.mark.parametrize("operation", ["stat", "read_bytes"]) @@ -143,6 +161,34 @@ def test_changed_tracks_content_and_ignores_identical_bytes_or_deletion( assert target.changed() is False +class TestDelegatedOutputScope: + def test_target_for_accepts_only_fresh_paths_inside_root(self, tmp_path: Path) -> None: + from autoskillit.core import DelegatedOutputScope + + root = tmp_path / "outputs" + root.mkdir() + preexisting = root / "old-plan.md" + preexisting.write_text("existing") + fresh = root / "new-plan.md" + outside = tmp_path / "outside.md" + scope = DelegatedOutputScope( + str(root.resolve()), + frozenset({str(preexisting.resolve())}), + ) + + assert scope.target_for(str(outside.resolve())) is None + assert scope.target_for(str(root.resolve())) is None + assert scope.target_for(str(preexisting.resolve())) is None + + target = scope.target_for(str(fresh.resolve())) + assert target == DelegatedWriteTarget(str(fresh.resolve()), None) + assert target is not None + assert target.changed() is False + + fresh.write_text("new plan") + assert target.changed() is True + + 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 8ef3152c01..48bd3c20a6 100644 --- a/tests/execution/conftest.py +++ b/tests/execution/conftest.py @@ -671,6 +671,7 @@ def make_write_evidence_scope(**overrides: Any) -> WriteEvidenceScope: "cwd": "", "skill_command": "", "delegated_target": None, + "delegated_output_scope": None, } values.update(overrides) return WriteEvidenceScope(**values) diff --git a/tests/execution/test_delegated_write_evidence.py b/tests/execution/test_delegated_write_evidence.py index 9344e5daf2..156971f726 100644 --- a/tests/execution/test_delegated_write_evidence.py +++ b/tests/execution/test_delegated_write_evidence.py @@ -56,14 +56,20 @@ def plan_and_target(tmp_path: Path) -> tuple[Path, DelegatedWriteTarget]: return plan, DelegatedWriteTarget.bind(os.path.realpath(plan)) -def _edit_stream(file_path: str, result_record: str | None = None) -> str: +def _edit_stream( + file_path: str, + result_record: str | None = None, + *, + tool_name: str = "Edit", + child_result_is_error: bool | 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_child_tool_use(file_path=file_path, tool_name=tool_name), + delegated_child_tool_result(is_error=child_result_is_error), delegated_task_notification(), delegated_root_agent_result(), delegated_root_text(), @@ -76,6 +82,262 @@ def _bound_behavior(target: DelegatedWriteTarget | None) -> WriteBehaviorSpec: return WriteBehaviorSpec(mode="always", delegated_target=target) +@pytest.mark.parametrize( + "scenario", + ["owned", "unbound", "failed", "nul", "worktree_temp"], + ids=["owned-success", "unbound", "failed-child", "nul-path", "worktree-temp-only"], +) +def test_collects_only_bound_successful_implementation_paths( + tmp_path: Path, scenario: str +) -> None: + from autoskillit.core import DelegatedOutputScope + from autoskillit.execution.session._write_evidence import _compute_write_evidence + from tests.execution.conftest import make_write_evidence_scope + + root = tmp_path.resolve() + child_path = root / "plan.md" + skill_command = "/autoskillit:prepare-pr" + write_watch_dirs: tuple[Path, ...] = () + if scenario == "worktree_temp": + child_path = root / ".autoskillit" / "temp" / "plan.md" + skill_command = "/autoskillit:implement-worktree plan" + write_watch_dirs = (root / "src",) + child_path_text = str(child_path) + if scenario == "nul": + child_path_text += "\x00invalid" + + output_scope = None if scenario == "unbound" else DelegatedOutputScope(str(root), frozenset()) + evidence_scope = make_write_evidence_scope( + cwd=str(root), + skill_command=skill_command, + write_watch_dirs=write_watch_dirs, + delegated_output_scope=output_scope, + ) + session = parse_session_result( + _edit_stream( + child_path_text, + tool_name="Write", + child_result_is_error=scenario == "failed", + ) + ) + + evidence = _compute_write_evidence(session, evidence_scope) + + expected = (os.path.realpath(child_path),) if scenario == "owned" else () + assert evidence.delegated_write_paths == expected + + +def test_canonical_path_rejects_nul( + tmp_path: Path, + plan_and_target: tuple[Path, DelegatedWriteTarget], +) -> None: + from autoskillit.execution.session._write_evidence import _compute_write_evidence + from tests.execution.conftest import make_write_evidence_scope + + plan, target = plan_and_target + plan.write_text(_PREPENDED_MARKER + plan.read_text()) + child_paths = (str(plan), f"{plan}\x00invalid") + records = [delegated_root_agent_call(), delegated_task_started(), delegated_child_prompt()] + for index, child_path in enumerate(child_paths): + tool_use_id = f"call_child_edit_{index}" + records.extend( + ( + delegated_child_tool_use( + tool_use_id=tool_use_id, + tool_name="Write", + file_path=child_path, + ), + delegated_child_tool_result(tool_use_id=tool_use_id, is_error=False), + ) + ) + records.extend( + (delegated_task_notification(), delegated_root_agent_result(), delegated_root_text()) + ) + records.append(delegated_result_record()) + session = parse_session_result(delegated_stream(records)) + evidence_scope = make_write_evidence_scope( + cwd=str(tmp_path), + delegated_target=target, + ) + + evidence = _compute_write_evidence(session, evidence_scope) + + assert evidence.delegated_write_count == 1 + + +@pytest.mark.parametrize("missing", ["output_name", "scope", "paths", "value"]) +def test_output_admission_returns_same_evidence_when_unarmed( + tmp_path: Path, + missing: str, +) -> None: + from autoskillit.core import DelegatedOutputScope, WriteEvidence + from autoskillit.execution.session._write_evidence import _admit_declared_output_write + + root = tmp_path.resolve() + output_path = root / "plan.md" + output_path.write_text("new output") + output_scope = DelegatedOutputScope(str(root), frozenset()) + observed_paths = () if missing == "paths" else (str(output_path),) + evidence = WriteEvidence( + write_call_count=0, + fs_writes_detected=False, + git_writes_detected=False, + delegated_write_paths=observed_paths, + ) + output_name = None if missing == "output_name" else "prep_path" + scope = None if missing == "scope" else output_scope + fields = {} if missing == "value" else {"prep_path": "plan.md"} + + admitted = _admit_declared_output_write(evidence, fields, output_name, scope, str(root)) + + assert admitted is evidence + + +@pytest.mark.parametrize("absolute", [False, True], ids=["relative-path", "absolute-path"]) +def test_output_admission_counts_matching_fresh_observations( + tmp_path: Path, + absolute: bool, +) -> None: + from autoskillit.core import DelegatedOutputScope, WriteEvidence + from autoskillit.execution.session._write_evidence import _admit_declared_output_write + + root = tmp_path.resolve() + output_path = root / "plan.md" + output_path.write_text("new output") + canonical_path = str(output_path.resolve()) + evidence = WriteEvidence( + write_call_count=2, + fs_writes_detected=True, + git_writes_detected=False, + file_changes_count=1, + delegated_write_count=3, + delegated_write_paths=(canonical_path, canonical_path, str(root / "other.md")), + ) + output_scope = DelegatedOutputScope(str(root), frozenset()) + emitted_path = str(output_path) if absolute else "plan.md" + + admitted = _admit_declared_output_write( + evidence, + {"prep_path": emitted_path}, + "prep_path", + output_scope, + str(root), + ) + + assert admitted is not evidence + assert admitted.write_call_count == evidence.write_call_count + assert admitted.file_changes_count == evidence.file_changes_count + assert admitted.delegated_write_count == evidence.delegated_write_count + 2 + assert admitted.delegated_write_paths == evidence.delegated_write_paths + + +@pytest.mark.parametrize("case", ["preexisting", "stale", "mismatched"]) +def test_output_admission_rejections_preserve_evidence_identity( + tmp_path: Path, + case: str, +) -> None: + from autoskillit.core import DelegatedOutputScope, WriteEvidence + from autoskillit.execution.session._write_evidence import _admit_declared_output_write + + root = tmp_path.resolve() + output_path = root / "plan.md" + canonical_path = str(output_path) + if case != "stale": + output_path.write_text("output") + preexisting = frozenset({canonical_path}) if case == "preexisting" else frozenset() + observed_path = str(root / "other.md") if case == "mismatched" else canonical_path + evidence = WriteEvidence( + write_call_count=1, + fs_writes_detected=False, + git_writes_detected=False, + delegated_write_paths=(observed_path,), + ) + output_scope = DelegatedOutputScope(str(root), preexisting) + + admitted = _admit_declared_output_write( + evidence, + {"prep_path": "plan.md"}, + "prep_path", + output_scope, + str(root), + ) + + assert admitted is evidence + + +def test_output_admission_preserves_identity_when_freshness_check_raises( + tmp_path: Path, + monkeypatch, +) -> None: + from autoskillit.core import DelegatedOutputScope, WriteEvidence + from autoskillit.execution.session._write_evidence import _admit_declared_output_write + + root = tmp_path.resolve() + output_path = root / "plan.md" + output_path.write_text("output") + evidence = WriteEvidence( + write_call_count=1, + fs_writes_detected=False, + git_writes_detected=False, + delegated_write_paths=(str(output_path),), + ) + output_scope = DelegatedOutputScope(str(root), frozenset()) + + def unavailable(_target: DelegatedWriteTarget) -> bool: + raise OSError("target unavailable") + + monkeypatch.setattr(DelegatedWriteTarget, "changed", unavailable) + with capture_logs() as events: + admitted = _admit_declared_output_write( + evidence, + {"prep_path": "plan.md"}, + "prep_path", + output_scope, + str(root), + ) + + assert admitted is evidence + assert len(events) == 1 + assert events[0]["event"] == "delegated_output_freshness_check_failed" + assert events[0]["output"] == "prep_path" + assert events[0]["error_type"] == "OSError" + assert events[0]["log_level"] == "debug" + + +def test_unarmed_conditional_output_does_not_log_admission_warning(tmp_path: Path) -> None: + from autoskillit.core import DelegatedOutputScope + + root = tmp_path.resolve() + behavior = WriteBehaviorSpec( + mode="conditional", + expected_when=(r"prep_path\s*=",), + delegated_output_scope=DelegatedOutputScope(str(root), frozenset()), + ) + contract = SimpleNamespace( + delegated_write_output="prep_path", + outputs=(), + outcome_invariants=(), + ) + + with capture_logs() as events: + result = _build_skill_result( + _make_result( + returncode=0, + stdout=_edit_stream(str(root / "plan.md"), tool_name="Write"), + ), + skill_command="/autoskillit:prepare-pr", + cwd=str(root), + write_behavior=behavior, + backend=ClaudeCodeBackend(), + skill_contract=contract, + ) + + assert result.success is True + assert result.evidence.delegated_write_paths == (os.path.realpath(root / "plan.md"),) + assert result.evidence.has_implementation_evidence is False + assert not any(event.get("event") == "delegated_output_write_not_admitted" for event in events) + + @pytest.mark.parametrize("fs_writes_detected", [False, True]) def test_bound_child_edit_is_implementation_progress( tmp_path: Path, diff --git a/tests/recipe/test_delegated_write_contracts.py b/tests/recipe/test_delegated_write_contracts.py index b8d12e52e7..578633e32e 100644 --- a/tests/recipe/test_delegated_write_contracts.py +++ b/tests/recipe/test_delegated_write_contracts.py @@ -16,6 +16,13 @@ def test_dry_walkthrough_declares_its_delegated_write_input() -> None: assert contract.delegated_write_input == "plan_path" +def test_prepare_pr_declares_its_delegated_write_output() -> None: + contract = get_skill_contract("prepare-pr", load_bundled_manifest()) + + assert contract is not None + assert contract.delegated_write_output == "prep_path" + + def test_every_bundled_skill_contract_parses() -> None: manifest = load_bundled_manifest() @@ -24,22 +31,44 @@ def test_every_bundled_skill_contract_parses() -> None: @pytest.mark.parametrize( - ("inputs", "delegated_input", "write_behavior"), + ("inputs", "outputs", "delegated_input", "delegated_output", "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"), + ([{"name": "plan_path", "type": "file_path"}], [], "unknown_path", None, "always"), + ([{"name": "plan_path", "type": "string"}], [], "plan_path", None, "always"), + ([{"name": "plan_path", "type": "file_path"}], [], "plan_path", None, None), + ([{"name": "plan_path", "type": "file_path"}], [], "plan_path", None, ["always"]), + ([{"name": "plan_path", "type": "file_path"}], [], 1, None, "always"), + ([], [{"name": "prep_path", "type": "file_path"}], None, 1, "always"), + ( + [], + [{"name": "prep_path", "type": "file_path"}], + None, + "unknown_path", + "always", + ), + ( + [{"name": "plan_path", "type": "file_path"}], + [], + None, + "plan_path", + "always", + ), + ([], [{"name": "prep_path", "type": "string"}], None, "prep_path", "always"), + ([], [{"name": "prep_path", "type": "file_path"}], None, "prep_path", None), ], ) def test_invalid_delegated_write_declarations_name_the_skill( inputs: list[dict[str, object]], + outputs: list[dict[str, object]], delegated_input: object, + delegated_output: object, write_behavior: object, ) -> None: - skill_data: dict[str, object] = {"inputs": inputs, "outputs": []} - skill_data["delegated_write_input"] = delegated_input + skill_data: dict[str, object] = {"inputs": inputs, "outputs": outputs} + if delegated_input is not None: + skill_data["delegated_write_input"] = delegated_input + if delegated_output is not None: + skill_data["delegated_write_output"] = delegated_output if write_behavior is not None: skill_data["write_behavior"] = write_behavior diff --git a/tests/server/test_run_skill_delegated_output_write.py b/tests/server/test_run_skill_delegated_output_write.py new file mode 100644 index 0000000000..1a42111815 --- /dev/null +++ b/tests/server/test_run_skill_delegated_output_write.py @@ -0,0 +1,516 @@ +"""A fresh declared output may receive credit for an owned child write.""" + +from __future__ import annotations + +import json +import os +from dataclasses import replace +from pathlib import Path +from types import SimpleNamespace +from typing import Any + +import pytest +from structlog.testing import capture_logs + +from tests.execution.conftest import ( + _mock_backend, + _result_ndjson, + _sr, + delegated_child_tool_result, + delegated_child_tool_use, + delegated_root_agent_call, + delegated_root_agent_result, +) + +pytestmark = [pytest.mark.layer("server"), pytest.mark.small] + +_PREP_CONTENT = "Prepared pull request notes.\n" + + +def _contract_and_spec(tmp_path: Path) -> tuple[Any, Any]: + from autoskillit.config import AutomationConfig + from autoskillit.recipe.contracts import get_skill_contract, load_bundled_manifest + from autoskillit.server._factory import make_context + from tests.server._factory_test_helpers import _runner + + contract = get_skill_contract("prepare-pr", load_bundled_manifest()) + assert contract is not None + ctx = make_context(AutomationConfig(), runner=_runner(), project_dir=tmp_path) + assert ctx.write_expected_resolver is not None + spec = ctx.write_expected_resolver("/autoskillit:prepare-pr") + assert spec is not None + return contract, spec + + +def _dispatch_state( + cwd: Path, + contract: Any, + spec: Any, + *, + skill_name: str = "prepare-pr", + output_dir: str = "", +) -> SimpleNamespace: + plan = cwd / "plans" / "test-plan.md" + plan.parent.mkdir(parents=True, exist_ok=True) + plan.write_text("# Test plan\n", encoding="utf-8") + inputs: tuple[tuple[str, object], ...] + if skill_name == "prepare-pr": + inputs = ( + ("plan_paths", [str(plan)]), + ("run_name", "test-run"), + ("base_branch", "main"), + ("closing_issue", ""), + ) + else: + inputs = (("plan_path", str(plan)),) + return SimpleNamespace( + skill_command=f"/autoskillit:{skill_name}", + _skill_contract=contract, + _stored_contract=None, + write_spec=spec, + _bound_recipe_inputs=inputs, + output_dir=output_dir, + _default_temp=cwd / ".autoskillit" / "temp" / "prepare-pr", + cwd=str(cwd), + ) + + +def _bound_state(cwd: Path, contract: Any, spec: Any) -> SimpleNamespace: + from autoskillit.server.tools.tools_execution._run_skill_session import ( + _bind_delegated_writes, + ) + + state = _dispatch_state(cwd, contract, spec) + _bind_delegated_writes(state) + return state + + +def _output_scope(spec: Any) -> Any | None: + assert hasattr(spec, "delegated_output_scope") + return getattr(spec, "delegated_output_scope") + + +def _assert_no_output_scope(spec: Any) -> None: + assert hasattr(spec, "delegated_output_scope") + assert getattr(spec, "delegated_output_scope") is None + + +def _prep_stream( + prep: str, + *, + agent_error: bool = False, + parent_id: str = "call_agent_A", + emitted: str | None = None, + child_path: str | None = None, + include_child: bool = True, +) -> str: + records = [delegated_root_agent_call()] + if include_child: + records.extend( + [ + delegated_child_tool_use( + tool_name="Write", + file_path=child_path or prep, + parent_tool_use_id=parent_id, + ), + delegated_child_tool_result(parent_tool_use_id=parent_id), + ] + ) + if agent_error: + records.append( + json.dumps( + { + "type": "user", + "message": { + "content": [ + { + "type": "tool_result", + "tool_use_id": "call_agent_A", + "is_error": True, + } + ] + }, + } + ) + ) + else: + records.append(delegated_root_agent_result()) + records.extend( + [ + json.dumps( + { + "type": "assistant", + "message": { + "content": [ + { + "type": "tool_use", + "name": "Read", + "id": "read_prep", + "input": {"file_path": prep}, + } + ] + }, + } + ), + json.dumps( + { + "type": "user", + "message": { + "content": [ + { + "type": "tool_result", + "tool_use_id": "read_prep", + "content": [{"type": "text", "text": _PREP_CONTENT}], + } + ] + }, + } + ), + _result_ndjson( + f"prep_path = {emitted or prep}\n" + "selected_lenses = none\n" + "lens_context_paths = none\n" + "%%ORDER_UP%%" + ), + ] + ) + return "\n".join(records) + + +def _root_write_stream(prep: str) -> str: + return "\n".join( + [ + delegated_child_tool_use( + tool_use_id="root_write", + tool_name="Write", + file_path=prep, + parent_tool_use_id=None, + agent_type=None, + ), + delegated_child_tool_result( + tool_use_id="root_write", + parent_tool_use_id=None, + agent_type=None, + ), + _result_ndjson( + f"prep_path = {prep}\n" + "selected_lenses = none\n" + "lens_context_paths = none\n" + "%%ORDER_UP%%" + ), + ] + ) + + +def _write_file(path: str | Path) -> None: + file_path = Path(path) + file_path.parent.mkdir(parents=True, exist_ok=True) + file_path.write_text(_PREP_CONTENT, encoding="utf-8") + + +def _skill_result(state: SimpleNamespace, contract: Any, stream: str) -> Any: + from autoskillit.execution.backends.claude import ClaudeCodeBackend + from autoskillit.execution.headless import _build_skill_result + + return _build_skill_result( + _sr(0, stdout=stream), + skill_command=state.skill_command, + cwd=state.cwd, + write_behavior=state.write_spec, + write_watch_dirs=(state._default_temp,), + skill_contract=contract, + backend=ClaudeCodeBackend(), + ) + + +@pytest.mark.anyio +async def test_fresh_child_written_prep_is_credited_end_to_end( + tmp_path: Path, minimal_ctx, monkeypatch: pytest.MonkeyPatch +) -> None: + from autoskillit.execution.headless import run_headless_core + from tests.conftest import _make_result + + contract, spec = _contract_and_spec(tmp_path) + state = _bound_state(tmp_path, contract, spec) + prep = state._default_temp / "pr_prep.md" + assert not prep.exists() + + async def runner(cmd, **kwargs): + if cmd and cmd[0] == "git": + return _make_result(returncode=1) + _write_file(prep) + return _make_result(returncode=0, stdout=_prep_stream(str(prep))) + + monkeypatch.setattr( + minimal_ctx, "backend", _mock_backend(pty_required=True, channel_b_capable=True) + ) + monkeypatch.setattr(minimal_ctx, "runner", runner) + result = await run_headless_core( + state.skill_command, + str(tmp_path), + minimal_ctx, + write_watch_dirs=(state._default_temp,), + write_behavior=state.write_spec, + skill_contract=contract, + ) + + assert result.success is True + assert result.subtype != "zero_writes" + assert result.evidence.write_call_count == 0 + assert result.evidence.delegated_write_count == 1 + + +@pytest.mark.parametrize( + "case", + [ + "denied-parent", + "unowned-child", + "wrong-output", + "absolute-output", + "traversal-output", + "symlink-output", + "preexisting-output", + "disappeared-output", + "filesystem-noise", + ], +) +def test_unproven_prep_write_stays_zero_writes(tmp_path: Path, case: str) -> None: + from autoskillit.server.tools.tools_execution._run_skill_session import ( + _bind_delegated_writes, + ) + + contract, spec = _contract_and_spec(tmp_path) + state = _dispatch_state(tmp_path, contract, spec) + root = state._default_temp + prep = str(root / "pr_prep.md") + child_path = prep + emitted = prep + include_child = True + + if case == "preexisting-output": + _write_file(prep) + _bind_delegated_writes(state) + + if case in {"denied-parent", "unowned-child", "preexisting-output", "disappeared-output"}: + _write_file(prep) + elif case == "wrong-output": + child_path = str(root / "pr_prep_A.md") + prep = str(root / "pr_prep_B.md") + emitted = prep + _write_file(child_path) + _write_file(prep) + elif case == "filesystem-noise": + include_child = False + _write_file(prep) + elif case.endswith("-output") and case != "preexisting-output": + root.mkdir(parents=True, exist_ok=True) + name = "pr_prep.md" + if case == "absolute-output": + prep = str(tmp_path / "outside" / name) + elif case == "traversal-output": + prep = str(root / ".." / "elsewhere" / name) + else: + outside = tmp_path / "outside" + outside.mkdir(parents=True, exist_ok=True) + (root / "alias").symlink_to(outside, target_is_directory=True) + prep = str(root / "alias" / name) + child_path = prep + emitted = prep + _write_file(prep) + + stream = _prep_stream( + prep, + agent_error=case == "denied-parent", + parent_id="unregistered-parent" if case == "unowned-child" else "call_agent_A", + emitted=emitted, + child_path=child_path, + include_child=include_child, + ) + if case == "disappeared-output": + Path(prep).unlink() + result = _skill_result(state, contract, stream) + + assert result.subtype == "zero_writes" + assert result.evidence.delegated_write_count == 0 + + +def test_root_written_prep_still_credited(tmp_path: Path) -> None: + contract, spec = _contract_and_spec(tmp_path) + state = _bound_state(tmp_path, contract, spec) + prep = state._default_temp / "pr_prep.md" + _write_file(prep) + + result = _skill_result(state, contract, _root_write_stream(str(prep))) + + assert result.success is True + assert result.evidence.write_call_count == 1 + + +def test_binder_records_preexisting_names(tmp_path: Path) -> None: + contract, spec = _contract_and_spec(tmp_path) + state = _dispatch_state(tmp_path, contract, spec) + state._default_temp.mkdir(parents=True) + (state._default_temp / "old.md").write_text("old", encoding="utf-8") + + from autoskillit.server.tools.tools_execution._run_skill_session import ( + _bind_delegated_writes, + ) + + _bind_delegated_writes(state) + scope = _output_scope(state.write_spec) + + assert scope is not None + assert scope.root == os.path.realpath(state._default_temp) + assert os.path.join(scope.root, "old.md") in scope.preexisting + + +def test_binder_missing_root_is_known_empty(tmp_path: Path) -> None: + contract, spec = _contract_and_spec(tmp_path) + state = _dispatch_state(tmp_path, contract, spec) + + from autoskillit.server.tools.tools_execution._run_skill_session import ( + _bind_delegated_writes, + ) + + _bind_delegated_writes(state) + scope = _output_scope(state.write_spec) + + assert scope is not None + assert scope.root == os.path.realpath(state._default_temp) + assert scope.preexisting == frozenset() + + +def test_binder_rejects_non_directory_ancestor(tmp_path: Path) -> None: + contract, spec = _contract_and_spec(tmp_path) + state = _dispatch_state(tmp_path, contract, spec) + temp_parent = state._default_temp.parent + temp_parent.parent.mkdir(parents=True) + temp_parent.write_text("not a directory", encoding="utf-8") + + from autoskillit.server.tools.tools_execution._run_skill_session import ( + _bind_delegated_writes, + ) + + with capture_logs() as events: + _bind_delegated_writes(state) + + _assert_no_output_scope(state.write_spec) + assert any(event.get("event") == "delegated_write_target_unbound" for event in events) + + +def test_binder_skips_explicit_output_dir(tmp_path: Path) -> None: + contract, spec = _contract_and_spec(tmp_path) + state = _dispatch_state(tmp_path, contract, spec, output_dir=".") + + from autoskillit.server.tools.tools_execution._run_skill_session import ( + _bind_delegated_writes, + ) + + with capture_logs() as events: + _bind_delegated_writes(state) + + _assert_no_output_scope(state.write_spec) + assert any( + event.get("event") == "delegated_write_target_unbound" + and event.get("reason") == "explicit_output_dir" + for event in events + ) + + +def test_binder_rejects_root_outside_cwd(tmp_path: Path) -> None: + contract, spec = _contract_and_spec(tmp_path) + cwd = tmp_path / "cwd" + cwd.mkdir() + state = _dispatch_state(cwd, contract, spec) + (cwd / ".autoskillit").mkdir() + outside = tmp_path / "outside" + outside.mkdir() + (cwd / ".autoskillit" / "temp").symlink_to(outside, target_is_directory=True) + + from autoskillit.server.tools.tools_execution._run_skill_session import ( + _bind_delegated_writes, + ) + + with capture_logs() as events: + _bind_delegated_writes(state) + + _assert_no_output_scope(state.write_spec) + assert any(event.get("event") == "delegated_write_target_unbound" for event in events) + + +def test_binder_rebinds_output_scope_on_resume(tmp_path: Path) -> None: + from autoskillit.core import DelegatedWriteTarget + from autoskillit.server.tools.tools_execution._run_skill_session import ( + _bind_delegated_writes, + ) + + contract, spec = _contract_and_spec(tmp_path) + state = _dispatch_state(tmp_path, contract, spec) + input_path = tmp_path / "input.md" + input_path.write_text("input", encoding="utf-8") + original_target = DelegatedWriteTarget.bind(str(input_path)) + state.write_spec = replace(state.write_spec, delegated_target=original_target) + state._stored_contract = object() + + _bind_delegated_writes(state) + + assert _output_scope(state.write_spec) is not None + assert state.write_spec.delegated_target is original_target + + +def test_binder_ignores_contracts_without_output_route(tmp_path: Path) -> None: + from autoskillit.config import AutomationConfig + from autoskillit.recipe.contracts import get_skill_contract, load_bundled_manifest + from autoskillit.server._factory import make_context + from autoskillit.server.tools.tools_execution._run_skill_session import ( + _bind_delegated_writes, + ) + from tests.server._factory_test_helpers import _runner + + contract = get_skill_contract("dry-walkthrough", load_bundled_manifest()) + assert contract is not None + ctx = make_context(AutomationConfig(), runner=_runner(), project_dir=tmp_path) + assert ctx.write_expected_resolver is not None + spec = ctx.write_expected_resolver("/autoskillit:dry-walkthrough") + assert spec is not None + state = _dispatch_state(tmp_path, contract, spec, skill_name="dry-walkthrough") + + _bind_delegated_writes(state) + + _assert_no_output_scope(state.write_spec) + assert state.write_spec.delegated_target is not None + + +def test_incomplete_baseline_stays_unbound_and_logs(tmp_path: Path, monkeypatch) -> None: + from autoskillit.core import TreeVanishedError + from autoskillit.server.tools.tools_execution import _run_skill_session + from autoskillit.server.tools.tools_execution._run_skill_session import ( + _bind_delegated_writes, + ) + + contract, spec = _contract_and_spec(tmp_path) + state = _dispatch_state(tmp_path, contract, spec) + state._default_temp.mkdir(parents=True) + + def incomplete_walk(root: Path): + raise TreeVanishedError("old.md", root) + + monkeypatch.setattr(_run_skill_session, "strict_walk", incomplete_walk, raising=False) + with capture_logs() as events: + _bind_delegated_writes(state) + + _assert_no_output_scope(state.write_spec) + assert any(event.get("event") == "delegated_write_target_unbound" for event in events) + + +def test_delegated_write_output_survives_contract_round_trip(tmp_path: Path) -> None: + from autoskillit.server.tools._execution_helpers import ( + deserialize_skill_contract, + serialize_skill_contract, + ) + + contract, _ = _contract_and_spec(tmp_path) + restored = deserialize_skill_contract(serialize_skill_contract(contract)) + + assert restored is not None + assert getattr(restored, "delegated_write_output", None) == "prep_path" diff --git a/tests/server/test_run_skill_delegated_write_binding.py b/tests/server/test_run_skill_delegated_write_binding.py index 62a2090141..9919d00ebc 100644 --- a/tests/server/test_run_skill_delegated_write_binding.py +++ b/tests/server/test_run_skill_delegated_write_binding.py @@ -17,7 +17,7 @@ serialize_skill_contract, ) from autoskillit.server.tools.tools_execution._run_skill_session import ( - _bind_delegated_target, + _bind_delegated_writes, ) pytestmark = [pytest.mark.layer("server"), pytest.mark.small] @@ -49,7 +49,7 @@ def test_binder_captures_in_scope_plan_digest(tmp_path: Path, path_form: str) -> }[path_form] state = _binding_state(root, (("plan_path", input_path),)) - _bind_delegated_target(state) + _bind_delegated_writes(state) target = state.write_spec.delegated_target assert target.path == str(plan.resolve()) @@ -79,7 +79,7 @@ def test_invalid_target_stays_unbound_and_logs_reason(tmp_path: Path, case: str) state = _binding_state(root, inputs[case]) with structlog.testing.capture_logs() as events: - _bind_delegated_target(state) + _bind_delegated_writes(state) assert state.write_spec.delegated_target is None assert any(event.get("event") == "delegated_write_target_unbound" for event in events) @@ -95,7 +95,7 @@ def unreadable(path: Path) -> bytes: monkeypatch.setattr(Path, "read_bytes", unreadable) with structlog.testing.capture_logs() as events: - _bind_delegated_target(state) + _bind_delegated_writes(state) assert state.write_spec.delegated_target is None assert any( @@ -110,7 +110,7 @@ def test_contract_without_declaration_leaves_write_spec_unchanged(tmp_path: Path state._skill_contract = SimpleNamespace(delegated_write_input=None) original_spec = state.write_spec - _bind_delegated_target(state) + _bind_delegated_writes(state) assert state.write_spec is original_spec @@ -152,7 +152,7 @@ def test_resume_keeps_original_target_baseline_without_rebinding(tmp_path: Path) WriteBehaviorSpec(mode="always", delegated_target=original_target), ) - _bind_delegated_target(state) + _bind_delegated_writes(state) restored_target = state.write_spec.delegated_target assert restored_target.baseline_digest == original_target.baseline_digest @@ -165,7 +165,7 @@ def test_resume_without_target_stays_unbound_without_warning(tmp_path: Path) -> state = _restored_state(tmp_path, contract, WriteBehaviorSpec(mode="always")) with structlog.testing.capture_logs() as events: - _bind_delegated_target(state) + _bind_delegated_writes(state) assert state.write_spec.delegated_target is None assert not any(event.get("event") == "delegated_write_target_unbound" for event in events)