diff --git a/src/autoskillit/core/__init__.pyi b/src/autoskillit/core/__init__.pyi index 6acedcaa42..ad284fca60 100644 --- a/src/autoskillit/core/__init__.pyi +++ b/src/autoskillit/core/__init__.pyi @@ -196,11 +196,18 @@ from .pipeline._step_context import current_step_name as current_step_name from .pipeline.pipeline_tracker import TrackerAuthorityReadResult as TrackerAuthorityReadResult from .pipeline.pipeline_tracker import TrackerAuthorityTarget as TrackerAuthorityTarget from .pipeline.pipeline_tracker import TrackerData as TrackerData +from .pipeline.pipeline_tracker import ( + TrackerDependencyAdmissionError as TrackerDependencyAdmissionError, +) +from .pipeline.pipeline_tracker import TrackerDependencyViolation as TrackerDependencyViolation +from .pipeline.pipeline_tracker import TrackerInitialization as TrackerInitialization from .pipeline.pipeline_tracker import TrackerMutation as TrackerMutation from .pipeline.pipeline_tracker import TrackerOwnerKind as TrackerOwnerKind from .pipeline.pipeline_tracker import TrackerParticipantKey as TrackerParticipantKey +from .pipeline.pipeline_tracker import dependency_cycle_members as dependency_cycle_members from .pipeline.pipeline_tracker import initialize_kitchen_tracker as initialize_kitchen_tracker from .pipeline.pipeline_tracker import initialize_manual_tracker as initialize_manual_tracker +from .pipeline.pipeline_tracker import is_tracker_enforced_step as is_tracker_enforced_step from .pipeline.pipeline_tracker import mutate_tracker as mutate_tracker from .pipeline.pipeline_tracker import pipeline_tracker_directory as pipeline_tracker_directory from .pipeline.pipeline_tracker import pipeline_tracker_path as pipeline_tracker_path diff --git a/src/autoskillit/core/pipeline/__init__.py b/src/autoskillit/core/pipeline/__init__.py index 77ce17d62b..4d21d7d868 100644 --- a/src/autoskillit/core/pipeline/__init__.py +++ b/src/autoskillit/core/pipeline/__init__.py @@ -16,9 +16,14 @@ from autoskillit.core.pipeline.pipeline_tracker import ( TrackerAuthorityReadResult, TrackerAuthorityTarget, + TrackerDependencyAdmissionError, + TrackerDependencyViolation, + TrackerInitialization, TrackerParticipantKey, + dependency_cycle_members, initialize_kitchen_tracker, initialize_manual_tracker, + is_tracker_enforced_step, mutate_tracker, pipeline_tracker_directory, pipeline_tracker_path, @@ -71,6 +76,9 @@ "OperationLeaseRecord", "TrackerAuthorityReadResult", "TrackerAuthorityTarget", + "TrackerDependencyAdmissionError", + "TrackerDependencyViolation", + "TrackerInitialization", "TrackerParticipantKey", "TurnSequence", "build_dfg", @@ -80,10 +88,12 @@ "current_operation_lease", "current_order_id", "current_step_name", + "dependency_cycle_members", "execution_marker", "filter_sessions_by_recipe", "format_top_bigrams", "is_parent_assistant_record", + "is_tracker_enforced_step", "initialize_kitchen_tracker", "initialize_manual_tracker", "iter_merged_assistant_turns", diff --git a/src/autoskillit/core/pipeline/pipeline_tracker.py b/src/autoskillit/core/pipeline/pipeline_tracker.py index 84ef0a75c1..86e631c4c2 100644 --- a/src/autoskillit/core/pipeline/pipeline_tracker.py +++ b/src/autoskillit/core/pipeline/pipeline_tracker.py @@ -10,10 +10,13 @@ import fcntl import json import os -from collections.abc import Callable, Mapping, MutableMapping +import uuid +from collections.abc import Callable, Iterable, Iterator, Mapping, MutableMapping, Sequence from copy import deepcopy -from dataclasses import dataclass +from dataclasses import dataclass, field +from datetime import UTC, datetime from pathlib import Path +from types import MappingProxyType from typing import Any, Literal import psutil @@ -29,14 +32,20 @@ ArtifactLease, acquire_flock_with_timeout, ) +from ..types import SKILL_TOOLS, FinalizedRecipeStep __all__ = [ "TrackerAuthorityReadResult", "TrackerAuthorityTarget", + "TrackerDependencyAdmissionError", + "TrackerDependencyViolation", "TrackerData", + "TrackerInitialization", "TrackerMutation", "TrackerOwnerKind", "TrackerParticipantKey", + "dependency_cycle_members", + "is_tracker_enforced_step", "initialize_kitchen_tracker", "initialize_manual_tracker", "mutate_tracker", @@ -53,6 +62,214 @@ TrackerData = dict[str, Any] TrackerMutation = Callable[[TrackerData], TrackerData] + +def is_tracker_enforced_step(step: object) -> bool: + """Only run_skill steps are dependency-gated and earn completion credit.""" + return isinstance(step, FinalizedRecipeStep) and step.tool in SKILL_TOOLS + + +def _dependency_cycle_edges( + graph: Mapping[str, Iterable[str]], +) -> set[tuple[str, str]]: + """Return only edges witnessed along DFS cycle paths and back edges.""" + white, gray, black = 0, 1, 2 + color: dict[str, int] = dict.fromkeys(graph, white) + cycle_edges: set[tuple[str, str]] = set() + + for root in graph: + if color[root] != white: + continue + color[root] = gray + path = [root] + frames: list[tuple[str, Iterator[str]]] = [(root, iter(graph[root]))] + while frames: + node, neighbors = frames[-1] + neighbor = next(neighbors, None) + if neighbor is None: + frames.pop() + path.pop() + color[node] = black + continue + if color.get(neighbor) == gray: + cycle_start = path.index(neighbor) + cycle_edges.update(zip(path[cycle_start:-1], path[cycle_start + 1 :])) + cycle_edges.add((node, neighbor)) + elif color.get(neighbor) == white: + color[neighbor] = gray + path.append(neighbor) + frames.append((neighbor, iter(graph.get(neighbor, ())))) + return cycle_edges + + +def dependency_cycle_members(graph: Mapping[str, Iterable[str]]) -> set[str]: + """Return the step names on witnessed cycle edges in *graph*.""" + return {name for edge in _dependency_cycle_edges(graph) for name in edge} + + +@dataclass(frozen=True, slots=True) +class TrackerDependencyViolation: + dependent: str + prerequisite: str + reason: Literal["unknown_step", "not_run_skill", "self_dependency", "cycle"] + detail: str + + +class TrackerDependencyAdmissionError(ValueError): + """A dependency graph contains edges the tracker cannot enforce.""" + + violations: tuple[TrackerDependencyViolation, ...] + + def __init__(self, violations: Sequence[TrackerDependencyViolation]) -> None: + self.violations = tuple(violations) + super().__init__("; ".join(violation.detail for violation in self.violations)) + + +def _normalize_tracker_dependencies( + dependencies: Mapping[str, Sequence[str]], +) -> dict[str, tuple[str, ...]]: + return { + dependent: tuple(dict.fromkeys(prerequisites)) + for dependent, prerequisites in dependencies.items() + if prerequisites + } + + +def _dependency_step_kind(step: object) -> str: + if not isinstance(step, FinalizedRecipeStep): + return "untyped" + return step.tool or step.action or "unconfigured" + + +def _dependency_endpoint_violation( + dependent: str, + prerequisite: str, + step_name: str, + role: str, + recipe_steps: Mapping[str, FinalizedRecipeStep], +) -> TrackerDependencyViolation | None: + if step_name not in recipe_steps: + return TrackerDependencyViolation( + dependent, + prerequisite, + "unknown_step", + f"{role} '{step_name}' is not a step in the finalized recipe", + ) + step = recipe_steps[step_name] + if is_tracker_enforced_step(step): + return None + kind = _dependency_step_kind(step) + article = "an" if kind == "unconfigured" else "a" + description = ( + f"{role} '{step_name}' is untyped" + if kind == "untyped" + else f"{role} '{step_name}' is {article} {kind} step" + ) + rule = "are dependency-gated" if role == "dependent" else "earn completion credit" + return TrackerDependencyViolation( + dependent, + prerequisite, + "not_run_skill", + f"{description}; only run_skill steps {rule}", + ) + + +def _dependency_endpoint_violations( + dependencies: Mapping[str, Sequence[str]], + recipe_steps: Mapping[str, FinalizedRecipeStep], +) -> tuple[list[TrackerDependencyViolation], dict[str, list[str]]]: + violations: list[TrackerDependencyViolation] = [] + admissible_graph: dict[str, list[str]] = {} + for dependent, prerequisites in dependencies.items(): + for prerequisite in prerequisites: + if dependent == prerequisite: + kind = ( + _dependency_step_kind(recipe_steps[dependent]) + if dependent in recipe_steps + else "unknown" + ) + violations.append( + TrackerDependencyViolation( + dependent, + prerequisite, + "self_dependency", + f"self-dependency: dependent/prerequisite '{dependent}' " + f"(kind: {kind}) " + "cannot be the same step", + ) + ) + continue + violation_count = len(violations) + for step_name, role in ( + (dependent, "dependent"), + (prerequisite, "prerequisite"), + ): + violation = _dependency_endpoint_violation( + dependent, + prerequisite, + step_name, + role, + recipe_steps, + ) + if violation is not None: + violations.append(violation) + if len(violations) == violation_count: + admissible_graph.setdefault(dependent, []).append(prerequisite) + admissible_graph.setdefault(prerequisite, []) + return violations, admissible_graph + + +def _dependency_cycle_violations( + graph: Mapping[str, Iterable[str]], +) -> list[TrackerDependencyViolation]: + return [ + TrackerDependencyViolation( + dependent, + prerequisite, + "cycle", + f"dependent '{dependent}' and prerequisite '{prerequisite}' " + "are run_skill steps in a cycle", + ) + for dependent, prerequisite in sorted(_dependency_cycle_edges(graph)) + ] + + +@dataclass(frozen=True, slots=True) +class TrackerInitialization: + kitchen_id: str + recipe_steps: Mapping[str, FinalizedRecipeStep] + dependencies: Mapping[str, Sequence[str]] = field(default_factory=dict) + skipped: frozenset[str] = frozenset() + + def __post_init__(self) -> None: + recipe_steps = dict(self.recipe_steps) + dependencies = _normalize_tracker_dependencies(self.dependencies) + endpoint_violations, admissible_graph = _dependency_endpoint_violations( + dependencies, recipe_steps + ) + violations = endpoint_violations + _dependency_cycle_violations(admissible_graph) + if violations: + raise TrackerDependencyAdmissionError(violations) + object.__setattr__(self, "recipe_steps", MappingProxyType(recipe_steps)) + object.__setattr__(self, "dependencies", MappingProxyType(dependencies)) + object.__setattr__(self, "skipped", frozenset(self.skipped)) + + def tracker_data(self, *, pipeline_id: str) -> TrackerData: + return { + "pipeline_id": pipeline_id, + "kitchen_id": self.kitchen_id, + "initialized_at": datetime.now(UTC).isoformat(), + "tracker_incarnation_id": uuid.uuid4().hex, + "steps": { + name: {"status": "skipped" if name in self.skipped else "pending"} + for name in self.recipe_steps + }, + "dependencies": { + dependent: list(prerequisites) + for dependent, prerequisites in self.dependencies.items() + }, + } + + _TRACKER_DIR_COMPONENTS = (".autoskillit", "temp", "pipeline_tracker") @@ -272,7 +489,9 @@ def _read_tracker_unlocked(target: TrackerAuthorityTarget) -> TrackerAuthorityRe target, error=( f"Expected pipeline tracker '{target.target_order_id}' is unavailable at " - f"{target.path}. Initialize or restore that exact tracker before retrying." + f"{target.path}. Create it with " + f"record_pipeline_step(op='init', pipeline_id='{target.target_order_id}') " + "(dependencies are optional) or restore that exact tracker, then retry." ), ) return TrackerAuthorityReadResult(target) @@ -310,11 +529,22 @@ def read_tracker_authority( def initialize_kitchen_tracker( target: TrackerAuthorityTarget, lease: ArtifactLease, - data: Mapping[str, Any], + initialization: TrackerInitialization, ) -> TrackerAuthorityReadResult: """Create or idempotently merge readable kitchen tracker authority.""" _require_shared_lease(target, lease) - incoming = _validate_tracker_data(dict(data)) + if not isinstance(initialization, TrackerInitialization): + raise TypeError( + f"initialization must be a TrackerInitialization; got {type(initialization).__name__}" + ) + if initialization.kitchen_id != target.target_order_id: + raise ValueError( + f"kitchen_id {initialization.kitchen_id!r} must match " + f"the tracker target id {target.target_order_id!r}" + ) + incoming = _validate_tracker_data( + initialization.tracker_data(pipeline_id=target.target_order_id) + ) with _TrackerLock(target): current = _read_tracker_unlocked(target) if current.error is not None: @@ -334,11 +564,17 @@ def initialize_kitchen_tracker( def initialize_manual_tracker( target: TrackerAuthorityTarget, lease: ArtifactLease, - data: Mapping[str, Any], + initialization: TrackerInitialization, ) -> TrackerAuthorityReadResult: """Create manual tracker authority, refusing every existing target.""" _require_shared_lease(target, lease) - incoming = _validate_tracker_data(dict(data)) + if not isinstance(initialization, TrackerInitialization): + raise TypeError( + f"initialization must be a TrackerInitialization; got {type(initialization).__name__}" + ) + incoming = _validate_tracker_data( + initialization.tracker_data(pipeline_id=target.target_order_id) + ) with _TrackerLock(target): if target.path.exists() or target.path.is_symlink(): return TrackerAuthorityReadResult( @@ -365,6 +601,8 @@ def mutate_tracker( if current.data is None: return current updated = _validate_tracker_data(mutation(deepcopy(current.data))) + if updated["dependencies"] != current.data["dependencies"]: + raise ValueError("tracker mutations must not change dependencies") if updated == current.data: return current atomic_write(target.path, json.dumps(updated)) diff --git a/src/autoskillit/server/tools/AGENTS.md b/src/autoskillit/server/tools/AGENTS.md index b22a5b8e73..31c6a48dbe 100644 --- a/src/autoskillit/server/tools/AGENTS.md +++ b/src/autoskillit/server/tools/AGENTS.md @@ -28,7 +28,7 @@ Files with no registration support their folder; files with several registration - `_commit_outcome.py` — Support module: Durable outcome accounting for commit_files responses. - `_ordering_telemetry.py` — Support module: Ordering-violation detection over pipeline session step records. - `_overlay_state.py` — Support module: Locked, validated access to the kitchen session overlay. -- `_pipeline_deps.py` — Support module: Curated Phase A dependencies from the finalized recipe routing graph. +- `_pipeline_deps.py` — Support module: Produces tracker dependencies from curated Phase A recipe edges and caller-requested edges. - `_preflight.py` — Support module: Dispatch-feasibility preflight — shared by open_kitchen and dispatch_food_truck. - `_pre_commit_failure.py` — Support module: Pre-commit failure classification helpers for commit_files. - `_pre_commit_transaction.py` — Support module: Pre-commit execution, re-stage, and retry transaction for commit_files. diff --git a/src/autoskillit/server/tools/_pipeline_deps.py b/src/autoskillit/server/tools/_pipeline_deps.py index c427a5f453..00240bb2f7 100644 --- a/src/autoskillit/server/tools/_pipeline_deps.py +++ b/src/autoskillit/server/tools/_pipeline_deps.py @@ -1,4 +1,4 @@ -"""Curated Phase A dependencies from the finalized recipe routing graph. +"""Produce tracker dependencies from recipe routes and caller requests. Underscore-prefixed helper module (not ``tools_*.py``) — exempt from the ``tools_*.py`` import-namespace restriction (REQ-ARCH-003 / REQ-IMP-003), @@ -10,7 +10,14 @@ from __future__ import annotations -from autoskillit.core import FinalizedRecipeProjection, get_logger +from collections.abc import Mapping, Sequence + +from autoskillit.core import ( + FinalizedRecipeProjection, + dependency_cycle_members, + get_logger, + is_tracker_enforced_step, +) logger = get_logger(__name__) @@ -27,34 +34,10 @@ } -def _find_cycle_members(graph: dict[str, set[str]]) -> set[str]: - """Return step names that participate in a cycle, via DFS back-edge detection.""" - WHITE, GRAY, BLACK = 0, 1, 2 - color: dict[str, int] = dict.fromkeys(graph, WHITE) - cycle_members: set[str] = set() - - def visit(node: str, stack: list[str]) -> None: - color[node] = GRAY - stack.append(node) - for neighbor in graph.get(node, ()): - if color.get(neighbor) == GRAY: - idx = stack.index(neighbor) - cycle_members.update(stack[idx:]) - elif color.get(neighbor) == WHITE: - visit(neighbor, stack) - stack.pop() - color[node] = BLACK - - for node in graph: - if color[node] == WHITE: - visit(node, []) - return cycle_members - - -def _derive_phase_a_deps( +def _curated_single_predecessors( projection: FinalizedRecipeProjection | None, ) -> dict[str, list[str]]: - """Derive curated, single-edge, loop-free step dependencies for Phase A. + """Select curated, single-edge, loop-free predecessors from the recipe graph. Steps involved in a cycle are skipped. Multiple matching predecessors are branch alternatives that the all-of tracker cannot represent, so they are @@ -68,7 +51,7 @@ def _derive_phase_a_deps( for edge in projection.ordered_flow_edges: if edge.target in step_names: graph[edge.source].add(edge.target) - cycle_members = _find_cycle_members(graph) + cycle_members = dependency_cycle_members(graph) predecessors: dict[str, set[str]] = {name: set() for name in graph} for name, successors in graph.items(): @@ -93,3 +76,46 @@ def _derive_phase_a_deps( predecessors=matched, ) return deps + + +def _derive_phase_a_deps( + projection: FinalizedRecipeProjection | None, +) -> dict[str, list[str]]: + """Derive curated Phase A dependencies that the tracker can enforce.""" + if projection is None: + return {} + + steps = {step.name: step for step in projection.ordered_steps} + dependencies: dict[str, list[str]] = {} + for dependent, prerequisites in _curated_single_predecessors(projection).items(): + for prerequisite in prerequisites: + if is_tracker_enforced_step(steps.get(dependent)) and is_tracker_enforced_step( + steps.get(prerequisite) + ): + dependencies[dependent] = [prerequisite] + else: + logger.warning( + "pipeline_dependency_unenforceable_step", + dependent=dependent, + prerequisite=prerequisite, + ) + return dependencies + + +def _tracker_dependencies( + projection: FinalizedRecipeProjection | None, + requested: Mapping[str, Sequence[str]] | None = None, +) -> dict[str, list[str]]: + """Combine enforceable recipe dependencies with distinct caller requests.""" + dependencies = { + dependent: list(prerequisites) + for dependent, prerequisites in _derive_phase_a_deps(projection).items() + } + for dependent, prerequisites in (requested or {}).items(): + if not prerequisites: + continue + stored = dependencies.setdefault(dependent, []) + for prerequisite in prerequisites: + if prerequisite not in stored: + stored.append(prerequisite) + return dependencies diff --git a/src/autoskillit/server/tools/tools_kitchen/_open_kitchen/_tracker_auto_init.py b/src/autoskillit/server/tools/tools_kitchen/_open_kitchen/_tracker_auto_init.py index c92646edcd..00d4ef04d4 100644 --- a/src/autoskillit/server/tools/tools_kitchen/_open_kitchen/_tracker_auto_init.py +++ b/src/autoskillit/server/tools/tools_kitchen/_open_kitchen/_tracker_auto_init.py @@ -6,11 +6,11 @@ from __future__ import annotations -from datetime import UTC, datetime from pathlib import Path from autoskillit.core import ( TrackerAuthorityTarget, + TrackerInitialization, get_logger, initialize_kitchen_tracker, pipeline_tracker_directory, @@ -25,7 +25,7 @@ _retain_kitchen_tracker_authority, try_retire_tracker, ) -from autoskillit.server.tools._pipeline_deps import _derive_phase_a_deps +from autoskillit.server.tools._pipeline_deps import _tracker_dependencies from autoskillit.server.tools.tools_kitchen._open_kitchen_errors import ( _kitchen_failure_envelope, ) @@ -65,7 +65,7 @@ def _auto_init_pipeline_tracker(tool_ctx: ToolContext) -> str | None: if not active_steps or projection is None: return None try: - deps = _derive_phase_a_deps(projection) + deps = _tracker_dependencies(projection) except Exception: logger.warning("pipeline_tracker_auto_init_deps_failed", exc_info=True) return None @@ -81,16 +81,13 @@ def _auto_init_pipeline_tracker(tool_ctx: ToolContext) -> str | None: return None key, lease = _retain_kitchen_tracker_authority(tool_ctx) - steps: dict[str, dict[str, str]] = {name: {"status": "pending"} for name in active_steps} - tracker_data = { - "kitchen_id": tool_ctx.kitchen_id, - "pipeline_id": tool_ctx.kitchen_id, - "steps": steps, - "dependencies": dict(deps), - "initialized_at": datetime.now(UTC).isoformat(), - } try: - result = initialize_kitchen_tracker(key.target, lease, tracker_data) + initialization = TrackerInitialization( + kitchen_id=tool_ctx.kitchen_id, + recipe_steps=active_steps, + dependencies=deps, + ) + result = initialize_kitchen_tracker(key.target, lease, initialization) except Exception: _release_kitchen_tracker_authority(tool_ctx, unregister=False, retire=False) raise diff --git a/src/autoskillit/server/tools/tools_pipeline_tracker/__init__.py b/src/autoskillit/server/tools/tools_pipeline_tracker/__init__.py index 8cb2894c46..2e02d76f85 100644 --- a/src/autoskillit/server/tools/tools_pipeline_tracker/__init__.py +++ b/src/autoskillit/server/tools/tools_pipeline_tracker/__init__.py @@ -25,7 +25,6 @@ recover_run_skill_result, ) from autoskillit.server.tools.tools_pipeline_tracker._status import ( - _build_tracker_steps, _compute_status_counts, ) @@ -35,7 +34,6 @@ "mark_step_skipped", "recover_run_skill_result", "record_pipeline_step", - "_build_tracker_steps", "_compute_status_counts", "_handle_complete", "_handle_init", diff --git a/src/autoskillit/server/tools/tools_pipeline_tracker/_handlers.py b/src/autoskillit/server/tools/tools_pipeline_tracker/_handlers.py index 72842d6650..b8cf7eb9e2 100644 --- a/src/autoskillit/server/tools/tools_pipeline_tracker/_handlers.py +++ b/src/autoskillit/server/tools/tools_pipeline_tracker/_handlers.py @@ -4,7 +4,6 @@ import json import os -import uuid from collections.abc import Mapping from datetime import UTC, datetime from pathlib import Path @@ -17,6 +16,8 @@ DISPATCH_ID_ENV_VAR, ArtifactLease, TrackerAuthorityTarget, + TrackerDependencyAdmissionError, + TrackerInitialization, get_logger, initialize_manual_tracker, mutate_tracker, @@ -43,9 +44,9 @@ ) from autoskillit.server.tools._cancellation_shield import _cancellation_shield from autoskillit.server.tools._overlay_state import read_overlay +from autoskillit.server.tools._pipeline_deps import _tracker_dependencies from autoskillit.server.tools._types import deny_envelope from autoskillit.server.tools.tools_pipeline_tracker._status import ( - _build_tracker_steps, _compute_status_counts, ) @@ -54,6 +55,7 @@ from autoskillit.pipeline import RunSkillCompletionReceipt, ToolContext logger = get_logger(__name__) +_MAX_REPORTED_DEPENDENCY_VIOLATIONS = 10 def _resolve_skipped_steps(project_dir: Path, pipeline_id: str) -> set[str]: @@ -65,6 +67,34 @@ def _resolve_skipped_steps(project_dir: Path, pipeline_id: str) -> set[str]: return set() +def _dependency_admission_denial(error: TrackerDependencyAdmissionError) -> str: + violations = error.violations + details = [ + f"{item.dependent} -> {item.prerequisite}: {item.detail}" + for item in violations[:_MAX_REPORTED_DEPENDENCY_VIOLATIONS] + ] + remaining = len(violations) - len(details) + message = "record_pipeline_step: dependencies were rejected; no tracker was created." + if details: + message += " " + "; ".join(details) + if remaining: + message += f" {remaining} additional violation(s) omitted." + recovery = ( + "Call record_pipeline_step(op='init') again without dependencies, or supply only " + "edges whose dependent and prerequisite are both run_skill steps of the open recipe. " + "The server already adds the recipe's own dependencies; other steps are never tracked " + "for completion." + ) + return json.dumps( + deny_envelope( + message, + stage="preflight:pipeline_tracker_dependencies", + retriable=False, + recovery=recovery, + ) + ) + + @mcp.tool(tags={"autoskillit", "kitchen", "kitchen-core"}, annotations={"readOnlyHint": True}) @session_scoped(SCOPE_ANY) @_cancellation_shield() @@ -77,9 +107,21 @@ async def record_pipeline_step( ) -> str: """Initialize, query, or mark completion on the pipeline step tracker. - **op="init"**: Creates the tracker file with the server-authoritative step list - from the currently open recipe. The LLM provides the dependency graph at init time. - Idempotent — calling init twice with the same pipeline_id returns an error. + **op="init"**: Creates the tracker for ``pipeline_id`` (default: the dispatch id) + from the open recipe's step list. Initialization is create-only; a second init for + the same id is refused. Call it once before the first step-named ``run_skill`` when + ``run_skill`` reports that its expected tracker is unavailable. ``dependencies`` is + optional and normally omitted because the server already adds the recipe's own + dependencies. A supplied mapping ``{dependent: [prerequisite, ...]}`` is added to + them; empty prerequisite lists are discarded, and both endpoints of every retained + edge must be ``run_skill`` steps of the open recipe. Other steps (clone, branch, + merge, ``run_cmd``, ``run_python``) are never tracked for completion, so an edge + that names one is rejected and no tracker is created. + + ``preflight:pipeline_tracker_dependencies`` rejects unsupported caller-supplied + edges and requires revising or omitting that mapping. ``pipeline_tracker:init`` + reports an initialization failure, including an exception while deriving the recipe + graph; diagnose the reported error or active recipe projection before retrying. **op="status"**: Returns the current state of all tracked steps. @@ -166,49 +208,50 @@ def _handle_init( dependencies: dict[str, list[str]] | None, ) -> str: active_steps = ctx.active_recipe_steps - if active_steps is None: + if not active_steps: return json.dumps( - { - "success": False, - "is_error": True, - "error": ( + deny_envelope( + ( "record_pipeline_step: no active recipe steps found. " "Ensure the kitchen is open with a loaded recipe." ), - } + stage="preflight:pipeline_tracker", + retriable=False, + ) ) - skipped = _resolve_skipped_steps(ctx.project_dir, target.target_order_id) - steps = _build_tracker_steps(active_steps, skipped) - - tracker_data = { - "pipeline_id": target.target_order_id, - "kitchen_id": ctx.kitchen_id, - "initialized_at": datetime.now(UTC).isoformat(), - "tracker_incarnation_id": uuid.uuid4().hex, - "steps": steps, - "dependencies": dependencies or {}, - } + try: + initialization = TrackerInitialization( + kitchen_id=ctx.kitchen_id, + recipe_steps=active_steps, + dependencies=_tracker_dependencies(ctx.active_recipe_projection, dependencies), + skipped=frozenset(_resolve_skipped_steps(ctx.project_dir, target.target_order_id)), + ) + except TrackerDependencyAdmissionError as exc: + return _dependency_admission_denial(exc) - authority = initialize_manual_tracker(target, lease, tracker_data) + authority = initialize_manual_tracker(target, lease, initialization) if authority.error is not None: return json.dumps( - { - "success": False, - "is_error": True, - "error": ( + deny_envelope( + ( f"record_pipeline_step: pipeline '{target.target_order_id}' " f"has already been initialized. {authority.error}" ), - } + stage="preflight:pipeline_tracker", + retriable=False, + ) ) + tracker = authority.data + assert tracker is not None return json.dumps( { "success": True, - "pipeline_id": target.target_order_id, - "step_count": len(steps), - "dependency_count": len(dependencies or {}), + "pipeline_id": tracker["pipeline_id"], + "step_count": len(tracker["steps"]), + "dependency_count": len(tracker["dependencies"]), + "dependencies": tracker["dependencies"], } ) diff --git a/src/autoskillit/server/tools/tools_pipeline_tracker/_status.py b/src/autoskillit/server/tools/tools_pipeline_tracker/_status.py index bb9e5719eb..52b3e55a21 100644 --- a/src/autoskillit/server/tools/tools_pipeline_tracker/_status.py +++ b/src/autoskillit/server/tools/tools_pipeline_tracker/_status.py @@ -2,17 +2,6 @@ from __future__ import annotations -from collections.abc import Mapping - - -def _build_tracker_steps( - active_steps: Mapping[str, object], skipped: set[str] -) -> dict[str, dict[str, str]]: - return { - name: {"status": "skipped"} if name in skipped else {"status": "pending"} - for name in active_steps - } - def _compute_status_counts( steps: dict[str, dict[str, object]], dependencies: dict[str, list[str]] diff --git a/tests/_test_filter.py b/tests/_test_filter.py index 763f1714a7..baab1025c1 100644 --- a/tests/_test_filter.py +++ b/tests/_test_filter.py @@ -1362,6 +1362,7 @@ class SourceMapRejection(enum.StrEnum): "server/test_open_kitchen_type_enforcement.py", "server/test_tools_kitchen_sous_chef.py", "server/test_open_kitchen_auto_init_tracker.py", + "server/test_record_pipeline_step_complete.py", "server/test_tools_execution_input_gates_cross_binding.py", "server/test_tools_execution_input_gates_real_contracts.py", # Skills file-level entry that exercises plan-set skill-contract wiring: diff --git a/tests/arch/test_tracker_admission_authority.py b/tests/arch/test_tracker_admission_authority.py new file mode 100644 index 0000000000..24263f0bef --- /dev/null +++ b/tests/arch/test_tracker_admission_authority.py @@ -0,0 +1,174 @@ +"""Pipeline tracker admission has a declared set of writers and callers.""" + +from __future__ import annotations + +import ast +from collections import Counter + +import pytest + +from tests.arch._helpers import SRC_ROOT + +pytestmark = [pytest.mark.medium] + +_EXPECTED_TRACKER_CALLS = ( + ( + "server/tools/tools_pipeline_tracker/_handlers.py", + "_handle_init", + "initialize_manual_tracker", + 1, + ), + ( + "server/tools/tools_kitchen/_open_kitchen/_tracker_auto_init.py", + "_auto_init_pipeline_tracker", + "initialize_kitchen_tracker", + 1, + ), + ( + "server/tools/tools_pipeline_tracker/_handlers.py", + "mark_step_complete", + "mutate_tracker", + 1, + ), + ( + "server/tools/tools_pipeline_tracker/_handlers.py", + "mark_step_skipped", + "mutate_tracker", + 1, + ), + ( + "server/tools/tools_pipeline_tracker/_handlers.py", + "_handle_complete", + "mark_step_complete", + 1, + ), + ( + "server/tools/tools_pipeline_tracker/_handlers.py", + "_apply_receipt_tracker_outcome", + "mark_step_complete", + 1, + ), + ( + "server/tools/tools_execution/_run_skill_admission.py", + "_admit_step_guard", + "mark_step_skipped", + 1, + ), + ( + "server/tools/tools_execution/_run_skill_admission.py", + "_handle_audit_reservation_decision", + "_begin_run_skill_completion", + 3, + ), + ( + "server/tools/tools_execution/_run_skill_dispatch.py", + "run_skill", + "_begin_run_skill_completion", + 1, + ), + ( + "server/tools/tools_execution/_run_skill_dispatch.py", + "_resolve_fresh_recipe_step", + "_check_pipeline_deps", + 1, + ), + ( + "server/tools/tools_execution/_run_skill_dispatch.py", + "run_skill", + "_check_pipeline_deps", + 1, + ), +) + +_TRACKED_CALLEES = frozenset(callee for _, _, callee, _ in _EXPECTED_TRACKER_CALLS) + + +def _terminal_callee(node: ast.Call) -> str | None: + if isinstance(node.func, ast.Name): + return node.func.id + if isinstance(node.func, ast.Attribute): + return node.func.attr + return None + + +class _TrackerCallVisitor(ast.NodeVisitor): + def __init__(self, path: str) -> None: + self.path = path + self.functions: list[str] = [] + self.calls: Counter[tuple[str, str, str]] = Counter() + + def _visit_function(self, node: ast.FunctionDef | ast.AsyncFunctionDef) -> None: + self.functions.append(self.functions[0] if self.functions else node.name) + self.generic_visit(node) + self.functions.pop() + + def visit_FunctionDef(self, node: ast.FunctionDef) -> None: + self._visit_function(node) + + def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None: + self._visit_function(node) + + def visit_Call(self, node: ast.Call) -> None: + callee = _terminal_callee(node) + if callee in _TRACKED_CALLEES: + function = self.functions[0] if self.functions else "" + self.calls[(self.path, function, callee)] += 1 + self.generic_visit(node) + + +def _tracker_call_inventory() -> Counter[tuple[str, str, str]]: + calls: Counter[tuple[str, str, str]] = Counter() + for path in sorted(SRC_ROOT.rglob("*.py")): + relative = path.relative_to(SRC_ROOT).as_posix() + visitor = _TrackerCallVisitor(relative) + visitor.visit(ast.parse(path.read_text(encoding="utf-8"), filename=str(path))) + calls.update(visitor.calls) + return calls + + +def test_tracker_writer_and_credit_callers_match_the_inventory() -> None: + expected = Counter( + { + (path, function, callee): count + for path, function, callee, count in _EXPECTED_TRACKER_CALLS + } + ) + actual = _tracker_call_inventory() + + assert actual == expected, ( + f"Tracker writer, credit, or gate callers changed.\n" + f"Missing calls: {dict(expected - actual)}\n" + f"Unexpected calls: {dict(actual - expected)}" + ) + seen = {callee for (_, _, callee), count in actual.items() if count} + assert _TRACKED_CALLEES <= seen + + +def test_tracker_writers_accept_only_an_admitted_initialization_type() -> None: + path = SRC_ROOT / "core" / "pipeline" / "pipeline_tracker.py" + module = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + writers = { + node.name: node + for node in ast.walk(module) + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) + and node.name in {"initialize_manual_tracker", "initialize_kitchen_tracker"} + } + + assert set(writers) == {"initialize_manual_tracker", "initialize_kitchen_tracker"} + for name, writer in writers.items(): + parameters = writer.args.args + assert len(parameters) >= 3, f"{name} must receive its tracker initialization" + annotation = parameters[2].annotation + assert isinstance(annotation, ast.Name) and annotation.id == "TrackerInitialization" + + +def test_dependency_gate_reads_step_kind_only_through_the_predicate() -> None: + path = SRC_ROOT / "server" / "tools" / "_pipeline_deps.py" + module = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + + assert not any( + isinstance(node, ast.Attribute) and node.attr == "tool" for node in ast.walk(module) + ) + assert not any( + isinstance(node, ast.Name) and node.id == "SKILL_TOOLS" for node in ast.walk(module) + ) diff --git a/tests/core/test_grab_bag_split_completeness.py b/tests/core/test_grab_bag_split_completeness.py index 1f36e350c2..2d964db1a9 100644 --- a/tests/core/test_grab_bag_split_completeness.py +++ b/tests/core/test_grab_bag_split_completeness.py @@ -522,6 +522,7 @@ def test_no_unintended_new_test_files_under_tests_core() -> None: "test_path_containment.py", "test_paths.py", "test_pipeline_tracker.py", + "test_pipeline_tracker_admission.py", "test_plugin_artifact_identity.py", "test__plugin_cache.py", "test_process_cleanup_result.py", diff --git a/tests/core/test_pipeline_tracker.py b/tests/core/test_pipeline_tracker.py index a21705eeda..0859266a52 100644 --- a/tests/core/test_pipeline_tracker.py +++ b/tests/core/test_pipeline_tracker.py @@ -2,6 +2,7 @@ import json import os +from collections.abc import Mapping, Sequence from datetime import UTC, datetime from typing import Literal @@ -13,8 +14,10 @@ from autoskillit.core import ( ArtifactLease, ArtifactLeaseContention, + FinalizedRecipeStep, TrackerAuthorityReadResult, TrackerAuthorityTarget, + TrackerInitialization, TrackerParticipantKey, initialize_kitchen_tracker, initialize_manual_tracker, @@ -58,6 +61,20 @@ def _tracker(*, kitchen_id: str = "kitchen-1"): } +def _initialization( + *step_names: str, + kitchen_id: str = "order-1", + dependencies: Mapping[str, Sequence[str]] | None = None, +) -> TrackerInitialization: + return TrackerInitialization( + kitchen_id=kitchen_id, + recipe_steps={ + name: FinalizedRecipeStep(name=name, tool="run_skill") for name in step_names + }, + dependencies={} if dependencies is None else dependencies, + ) + + def _registry(monkeypatch, tmp_path, payload: object | None) -> None: registry_path = tmp_path / "registry" / "active_kitchens.json" lock_path = tmp_path / "registry" / "active_kitchens.lock" @@ -158,11 +175,12 @@ def test_locked_initialization_and_mutation_preserve_invalid_existing_bytes(tmp_ target.path.parent.mkdir(parents=True, exist_ok=True) target.path.write_bytes(b"{broken") try: - kitchen_result = initialize_kitchen_tracker(target, lease, _tracker()) + initialization = _initialization("prepare", kitchen_id=target.target_order_id) + kitchen_result = initialize_kitchen_tracker(target, lease, initialization) assert kitchen_result.error is not None assert target.path.read_bytes() == b"{broken" - manual_result = initialize_manual_tracker(target, lease, _tracker()) + manual_result = initialize_manual_tracker(target, lease, initialization) assert "create-only" in (manual_result.error or "") assert target.path.read_bytes() == b"{broken" @@ -232,12 +250,17 @@ def test_kitchen_merge_preserves_progress_and_manual_init_is_create_only(tmp_pat leases = {} lease = retain_tracker_lease(leases, _key(target, tmp_path, "manual")) try: - initial = _tracker() - assert initialize_manual_tracker(target, lease, initial).data == initial + initial = _initialization("prepare", kitchen_id=target.target_order_id) + created = initialize_manual_tracker(target, lease, initial) + assert created.data is not None + assert created.data["pipeline_id"] == target.target_order_id + assert created.data["kitchen_id"] == target.target_order_id + assert created.data["steps"] == {"prepare": {"status": "pending"}} + assert created.data["dependencies"] == {} + assert created.data["tracker_incarnation_id"] assert initialize_manual_tracker(target, lease, initial).error is not None - incoming = _tracker() - incoming["steps"]["new"] = {"status": "pending"} + incoming = _initialization("new", "prepare", kitchen_id=target.target_order_id) current = json.loads(target.path.read_text()) current["steps"]["prepare"] = {"status": "complete"} target.path.write_text(json.dumps(current)) @@ -255,16 +278,26 @@ def test_kitchen_merge_recomputes_dependencies_and_preserves_progress(tmp_path): target = _target(tmp_path, expected=False) lease = retain_tracker_lease({}, _key(target, tmp_path)) try: - initial = _tracker() - initial["dependencies"] = {"dry_walkthrough": ["make_plan"]} - assert initialize_kitchen_tracker(target, lease, initial).data == initial + steps = ("prepare", "dry_walkthrough", "make_plan") + initial = _initialization( + *steps, + kitchen_id=target.target_order_id, + dependencies={"dry_walkthrough": ["make_plan"]}, + ) + created = initialize_kitchen_tracker(target, lease, initial) + assert created.data is not None + assert created.data["steps"] == { + "prepare": {"status": "pending"}, + "dry_walkthrough": {"status": "pending"}, + "make_plan": {"status": "pending"}, + } + assert created.data["dependencies"] == {"dry_walkthrough": ["make_plan"]} current = json.loads(target.path.read_text()) current["steps"]["prepare"] = {"status": "complete"} target.path.write_text(json.dumps(current)) - refreshed = _tracker() - refreshed["dependencies"] = {} + refreshed = _initialization(*steps, kitchen_id=target.target_order_id) result = initialize_kitchen_tracker(target, lease, refreshed) assert result.data is not None diff --git a/tests/core/test_pipeline_tracker_admission.py b/tests/core/test_pipeline_tracker_admission.py new file mode 100644 index 0000000000..159e6c1078 --- /dev/null +++ b/tests/core/test_pipeline_tracker_admission.py @@ -0,0 +1,347 @@ +"""Admission snapshots keep tracker dependency edges tied to executable steps.""" + +from __future__ import annotations + +import json +from collections.abc import Mapping, Sequence + +import pytest + +from autoskillit.core.pipeline.pipeline_tracker import ( + TrackerDependencyAdmissionError, + TrackerInitialization, + dependency_cycle_members, + initialize_kitchen_tracker, + initialize_manual_tracker, + is_tracker_enforced_step, + mutate_tracker, + read_tracker_authority, + retain_tracker_lease, +) +from autoskillit.core.types import SKILL_TOOLS, FinalizedRecipeStep +from tests.core.test_pipeline_tracker import _key, _target, _tracker + +pytestmark = [pytest.mark.layer("core"), pytest.mark.small] + + +def _steps() -> dict[str, FinalizedRecipeStep]: + return { + "plan": FinalizedRecipeStep(name="plan", tool="run_skill"), + "verify": FinalizedRecipeStep(name="verify", tool="run_skill"), + "rectify": FinalizedRecipeStep(name="rectify", tool="run_skill"), + "review_approach": FinalizedRecipeStep(name="review_approach", tool="run_skill"), + "clone": FinalizedRecipeStep(name="clone", tool="bootstrap_clone"), + "done": FinalizedRecipeStep(name="done", action="stop"), + } + + +def _skill_steps(*names: str) -> dict[str, FinalizedRecipeStep]: + return {name: FinalizedRecipeStep(name=name, tool="run_skill") for name in names} + + +def _initialization( + *, + kitchen_id: str = "kitchen-1", + recipe_steps: Mapping[str, FinalizedRecipeStep] | None = None, + dependencies: Mapping[str, Sequence[str]] | None = None, + skipped: frozenset[str] | set[str] = frozenset(), +) -> TrackerInitialization: + return TrackerInitialization( + kitchen_id=kitchen_id, + recipe_steps=_steps() if recipe_steps is None else recipe_steps, + dependencies={} if dependencies is None else dependencies, + skipped=skipped, + ) + + +@pytest.mark.parametrize( + ("dependencies", "reason", "detail_parts"), + [ + ( + {"plan": ["clone"]}, + "not_run_skill", + ("prerequisite 'clone'", "bootstrap_clone"), + ), + ( + {"clone": ["plan"]}, + "not_run_skill", + ("dependent 'clone'", "bootstrap_clone"), + ), + ({"plan": ["done"]}, "not_run_skill", ("prerequisite 'done'", "stop")), + ({"plan": ["ghost"]}, "unknown_step", ("ghost",)), + ({"ghost": ["plan"]}, "unknown_step", ("ghost",)), + ({"plan": ["plan"]}, "self_dependency", ("plan",)), + ], +) +def test_admission_rejects_invalid_dependency_edges(dependencies, reason, detail_parts): + with pytest.raises(TrackerDependencyAdmissionError) as raised: + _initialization(dependencies=dependencies) + + violations = raised.value.violations + assert any( + violation.reason == reason and all(part in violation.detail for part in detail_parts) + for violation in violations + ) + if dependencies == {"plan": ["clone"]}: + assert len(violations) == 1 + + +def test_admission_reports_cycle_edges_without_marking_a_connecting_edge(): + cycle_edges = { + ("a", "b"), + ("b", "a"), + ("c", "d"), + ("d", "c"), + } + with pytest.raises(TrackerDependencyAdmissionError) as raised: + _initialization( + recipe_steps=_skill_steps("a", "b", "c", "d"), + dependencies={"a": ["b"], "b": ["a", "c"], "c": ["d"], "d": ["c"]}, + ) + + reported = { + (violation.dependent, violation.prerequisite) + for violation in raised.value.violations + if violation.reason == "cycle" + } + assert reported + assert reported <= cycle_edges + assert ("b", "c") not in reported + + +@pytest.mark.parametrize( + ("dependencies", "cycle_edges"), + [ + ( + {"plan": ["verify"], "verify": ["plan"]}, + {("plan", "verify"), ("verify", "plan")}, + ), + ( + {"plan": ["verify"], "verify": ["rectify"], "rectify": ["plan"]}, + {("plan", "verify"), ("verify", "rectify"), ("rectify", "plan")}, + ), + ], +) +def test_admission_rejects_cycles_and_names_a_ring_edge(dependencies, cycle_edges): + with pytest.raises(TrackerDependencyAdmissionError) as raised: + _initialization(dependencies=dependencies) + + reported = { + (violation.dependent, violation.prerequisite) + for violation in raised.value.violations + if violation.reason == "cycle" + } + assert reported & cycle_edges + + +@pytest.mark.parametrize("cyclic", [False, True]) +def test_admission_handles_deep_dependency_chains(cyclic): + names = tuple(f"step-{index}" for index in range(1500)) + dependencies = {left: [right] for left, right in zip(names, names[1:])} + steps = _skill_steps(*names) + if cyclic: + dependencies[names[-1]] = [names[-2]] + with pytest.raises(TrackerDependencyAdmissionError) as raised: + _initialization(recipe_steps=steps, dependencies=dependencies) + reported = { + (violation.dependent, violation.prerequisite) + for violation in raised.value.violations + if violation.reason == "cycle" + } + assert reported == {(names[-2], names[-1]), (names[-1], names[-2])} + assert dependency_cycle_members(dependencies) == set(names[-2:]) + else: + admitted = _initialization(recipe_steps=steps, dependencies=dependencies) + assert dict(admitted.dependencies) == { + name: tuple(prerequisites) for name, prerequisites in dependencies.items() + } + assert dependency_cycle_members(dependencies) == set() + + +def test_admission_describes_unconfigured_steps(): + with pytest.raises( + TrackerDependencyAdmissionError, + match="prerequisite 'empty' is an unconfigured step", + ): + _initialization( + recipe_steps=_skill_steps("plan") | {"empty": FinalizedRecipeStep(name="empty")}, + dependencies={"plan": ["empty"]}, + ) + + +def test_admission_rejects_edges_with_untyped_step_values(): + with pytest.raises(TrackerDependencyAdmissionError) as raised: + _initialization( + recipe_steps={"left": {}, "right": {}}, + dependencies={"left": ["right"]}, + ) + + assert any( + violation.reason == "not_run_skill" and "untyped" in violation.detail + for violation in raised.value.violations + ) + + +def test_admission_normalizes_empty_and_duplicate_prerequisites(): + empty = _initialization(dependencies={"verify": []}) + duplicate = _initialization(dependencies={"verify": ["plan", "plan"]}) + + assert dict(empty.dependencies) == {} + assert dict(duplicate.dependencies) == {"verify": ("plan",)} + + +def test_admission_allows_valid_and_untyped_steps_without_edges(): + accepted = _initialization(dependencies={"verify": ["plan"]}) + untyped = _initialization(recipe_steps={"opaque": {}}) + + assert dict(accepted.dependencies) == {"verify": ("plan",)} + assert dict(untyped.dependencies) == {} + + +def test_tracker_data_and_admitted_inputs_are_snapshots(): + recipe_steps = _steps() + prerequisites = ["plan"] + dependencies = {"verify": prerequisites} + skipped = {"plan"} + initialization = _initialization( + recipe_steps=recipe_steps, + dependencies=dependencies, + skipped=skipped, + ) + + recipe_steps["later"] = FinalizedRecipeStep(name="later", tool="run_skill") + del recipe_steps["plan"] + prerequisites[0] = "rectify" + dependencies["rectify"] = ["review_approach"] + skipped.clear() + skipped.add("verify") + data = initialization.tracker_data(pipeline_id="p") + + assert isinstance(data["tracker_incarnation_id"], str) + assert data["tracker_incarnation_id"] + assert data["pipeline_id"] == "p" + assert data["kitchen_id"] == "kitchen-1" + assert data["initialized_at"] + assert data["steps"] == { + name: {"status": "skipped" if name == "plan" else "pending"} for name in _steps() + } + assert data["dependencies"] == {"verify": ["plan"]} + + +@pytest.mark.parametrize( + ("initializer", "owner_kind"), + [(initialize_manual_tracker, "manual"), (initialize_kitchen_tracker, "kitchen")], +) +def test_tracker_writers_require_admitted_initialization(tmp_path, initializer, owner_kind): + target = _target(tmp_path, "p", expected=False) + lease = retain_tracker_lease({}, _key(target, tmp_path, owner_kind)) + try: + with pytest.raises(TypeError, match="TrackerInitialization; got dict"): + initializer(target, lease, _tracker()) + assert not target.path.exists() + finally: + lease.close() + + +def test_kitchen_reinitialization_preserves_identity_and_progress(tmp_path): + target = _target(tmp_path, "p", expected=False) + lease = retain_tracker_lease({}, _key(target, tmp_path, "kitchen")) + first = _initialization(kitchen_id="p", dependencies={"verify": ["plan"]}) + second = _initialization(kitchen_id="p", dependencies={"rectify": ["verify"]}) + try: + created = initialize_kitchen_tracker(target, lease, first) + + def complete_plan(data): + data["steps"]["plan"]["status"] = "complete" + return data + + mutate_tracker(target, lease, complete_plan) + refreshed = initialize_kitchen_tracker(target, lease, second) + finally: + lease.close() + + assert created.data is not None + assert refreshed.data is not None + assert refreshed.data["tracker_incarnation_id"] == created.data["tracker_incarnation_id"] + assert refreshed.data["steps"]["plan"] == {"status": "complete"} + assert refreshed.data["dependencies"] == {"rectify": ["verify"]} + + +def test_kitchen_reinitialization_adds_identity_to_legacy_tracker(tmp_path): + target = _target(tmp_path, "p", expected=False) + lease = retain_tracker_lease({}, _key(target, tmp_path, "kitchen")) + legacy = _tracker() + legacy["pipeline_id"] = "p" + legacy["kitchen_id"] = "p" + target.path.parent.mkdir(parents=True, exist_ok=True) + target.path.write_text(json.dumps(legacy)) + initialization = _initialization( + kitchen_id="p", + recipe_steps={"prepare": FinalizedRecipeStep(name="prepare", tool="run_skill")}, + ) + try: + result = initialize_kitchen_tracker(target, lease, initialization) + finally: + lease.close() + + assert result.data is not None + assert isinstance(result.data.get("tracker_incarnation_id"), str) + assert result.data["tracker_incarnation_id"] + + +def test_kitchen_initialization_rejects_a_different_target_id(tmp_path): + target = _target(tmp_path, "p", expected=False) + lease = retain_tracker_lease({}, _key(target, tmp_path, "kitchen")) + try: + with pytest.raises(ValueError, match="kitchen_id 'other-kitchen'.*target id 'p'"): + initialize_kitchen_tracker( + target, + lease, + _initialization(kitchen_id="other-kitchen"), + ) + assert not target.path.exists() + finally: + lease.close() + + +def test_mutation_cannot_change_dependencies(tmp_path): + target = _target(tmp_path, "p", expected=False) + lease = retain_tracker_lease({}, _key(target, tmp_path, "kitchen")) + initialization = _initialization(kitchen_id="p", dependencies={"verify": ["plan"]}) + try: + initialize_kitchen_tracker(target, lease, initialization) + original = target.path.read_bytes() + + def change_dependencies(data): + data["dependencies"] = {"rectify": ["verify"]} + return data + + with pytest.raises(ValueError): + mutate_tracker(target, lease, change_dependencies) + assert target.path.read_bytes() == original + finally: + lease.close() + + +def test_expected_absent_tracker_error_guides_initialization(tmp_path): + target = _target(tmp_path, "missing-pipeline", expected=True) + lease = retain_tracker_lease({}, _key(target, tmp_path)) + try: + result = read_tracker_authority(target, lease) + finally: + lease.close() + + assert result.error is not None + assert "record_pipeline_step(op='init'" in result.error + assert "missing-pipeline" in result.error + + +def test_tracker_enforced_steps_are_skill_tools_only(): + assert SKILL_TOOLS + for tool in SKILL_TOOLS: + assert is_tracker_enforced_step(FinalizedRecipeStep(name="skill", tool=tool)) + assert not is_tracker_enforced_step(FinalizedRecipeStep(name="clone", tool="bootstrap_clone")) + assert not is_tracker_enforced_step(FinalizedRecipeStep(name="python", tool="run_python")) + assert not is_tracker_enforced_step(FinalizedRecipeStep(name="done", action="stop")) + assert not is_tracker_enforced_step({"tool": "run_skill"}) + assert not is_tracker_enforced_step(None) diff --git a/tests/infra/test_schema_version_convention.py b/tests/infra/test_schema_version_convention.py index 460c17184d..4339591e8f 100644 --- a/tests/infra/test_schema_version_convention.py +++ b/tests/infra/test_schema_version_convention.py @@ -130,9 +130,9 @@ def _is_yaml_dump(node: ast.expr) -> bool: ("src/autoskillit/server/tools/_overlay_state.py", 132), # pipeline_tracker.py — locked tracker init/merge/mutation writes centralize the existing # unversioned tracker file schema. - ("src/autoskillit/core/pipeline/pipeline_tracker.py", 330), - ("src/autoskillit/core/pipeline/pipeline_tracker.py", 352), - ("src/autoskillit/core/pipeline/pipeline_tracker.py", 370), + ("src/autoskillit/core/pipeline/pipeline_tracker.py", 560), + ("src/autoskillit/core/pipeline/pipeline_tracker.py", 588), + ("src/autoskillit/core/pipeline/pipeline_tracker.py", 608), # tools_status.py — mcp_data dict ("src/autoskillit/server/tools/tools_status.py", 592), # tools_github.py — bug report dict (non-blocking report-bug status file) diff --git a/tests/server/_pipeline_test_helpers.py b/tests/server/_pipeline_test_helpers.py index 3525b8bf55..fbaa6f8a9e 100644 --- a/tests/server/_pipeline_test_helpers.py +++ b/tests/server/_pipeline_test_helpers.py @@ -35,8 +35,8 @@ def _setup_project(tmp_path, tool_ctx_kitchen_open, active_recipe_steps=None): tool_ctx_kitchen_open.project_dir = tmp_path if active_recipe_steps is None: active_recipe_steps = { - "rectify": RecipeStep(name="rectify", on_success="review_approach"), - "review_approach": RecipeStep(name="review_approach"), + "rectify": RecipeStep(name="rectify", tool="run_skill", on_success="review_approach"), + "review_approach": RecipeStep(name="review_approach", tool="run_skill"), } projection = _make_finalized_projection_from_recipe_steps(active_recipe_steps) tool_ctx_kitchen_open.active_recipe_projection = projection diff --git a/tests/server/test_complete_run_skill_result.py b/tests/server/test_complete_run_skill_result.py index 25c9f1c266..3c487d495d 100644 --- a/tests/server/test_complete_run_skill_result.py +++ b/tests/server/test_complete_run_skill_result.py @@ -13,6 +13,69 @@ class TestCompleteRunSkillResult: + @pytest.mark.anyio + async def test_unknown_step_receipt_tracker_parity(self, tool_ctx_kitchen_open, tmp_path): + from types import SimpleNamespace + + from autoskillit.server.tools.tools_kitchen._open_kitchen._tracker_auto_init import ( + _auto_init_pipeline_tracker, + ) + from autoskillit.server.tools.tools_pipeline_tracker import record_pipeline_step + from tests.server._pipeline_test_helpers import _setup_project + + _setup_project(tmp_path, tool_ctx_kitchen_open) + assert _auto_init_pipeline_tracker(tool_ctx_kitchen_open) is None + kitchen_target = tool_ctx_kitchen_open.kitchen_tracker_key.target + kitchen_tracker = json.loads(kitchen_target.path.read_text()) + assert isinstance(kitchen_tracker["tracker_incarnation_id"], str) + assert kitchen_tracker["tracker_incarnation_id"] + + manual_pipeline_id = "manual-receipt-tracker" + initialized = json.loads( + await record_pipeline_step(pipeline_id=manual_pipeline_id, op="init") + ) + assert initialized["success"] is True + manual_path = ( + tmp_path / ".autoskillit" / "temp" / "pipeline_tracker" / f"{manual_pipeline_id}.json" + ) + manual_tracker = json.loads(manual_path.read_text()) + + tracker_cases = ( + ( + kitchen_target.target_order_id, + kitchen_target.path, + kitchen_tracker, + ), + (manual_pipeline_id, manual_path, manual_tracker), + ) + results = [] + for pipeline_id, tracker_path, tracker in tracker_cases: + receipt = _publish_success_receipt( + tool_ctx_kitchen_open, + pipeline_id=pipeline_id, + tracker_path=tracker_path, + tracker_kitchen_id=tracker["kitchen_id"], + tracker_incarnation_id=tracker["tracker_incarnation_id"], + step_name="unknown-tracker-step", + ) + before = tracker_path.read_bytes() + result = json.loads( + await complete_run_skill_result( + receipt.receipt_id, + ctx=SimpleNamespace(session_id="request-session"), + ) + ) + after = tracker_path.read_bytes() + + assert before == after + results.append(result) + + assert results[0]["tracker"] == results[1]["tracker"] + assert results[0]["tracker"]["success"] is False + assert results[0]["tracker"]["stage"] == "mark_step_complete" + assert results[0]["tracker_repairable"] is results[1]["tracker_repairable"] + assert results[0]["tracker_repairable"] is True + @pytest.mark.anyio async def test_tracker_preparation_failure_keeps_acknowledged_credit_repairable( self, tool_ctx_kitchen_open, monkeypatch, tmp_path diff --git a/tests/server/test_open_kitchen_auto_init_tracker.py b/tests/server/test_open_kitchen_auto_init_tracker.py index 008052ded1..bf7db1ac24 100644 --- a/tests/server/test_open_kitchen_auto_init_tracker.py +++ b/tests/server/test_open_kitchen_auto_init_tracker.py @@ -20,7 +20,10 @@ def _configure_open_kitchen_mock(ctx, steps, tmp_path): from unittest.mock import MagicMock + from autoskillit.execution.backends import get_backend + ctx.project_dir = tmp_path + ctx.backend = get_backend(ctx.config.agent_backend.backend) ctx.recipes.load_and_validate.return_value = _with_finalized_projection( { "content": "name: remediation\nsteps: {}\n", @@ -101,8 +104,10 @@ def test_auto_init_releases_authority_when_initialization_raises(self, monkeypat _install_active_recipe_projection( ctx, { - "rectify": RecipeStep(name="rectify", on_success="review_approach"), - "review_approach": RecipeStep(name="review_approach"), + "rectify": RecipeStep( + name="rectify", tool="run_skill", on_success="review_approach" + ), + "review_approach": RecipeStep(name="review_approach", tool="run_skill"), }, ) @@ -130,8 +135,10 @@ def test_auto_init_preserves_corrupt_authority(self, tmp_path): _install_active_recipe_projection( ctx, { - "rectify": RecipeStep(name="rectify", on_success="review_approach"), - "review_approach": RecipeStep(name="review_approach"), + "rectify": RecipeStep( + name="rectify", tool="run_skill", on_success="review_approach" + ), + "review_approach": RecipeStep(name="review_approach", tool="run_skill"), }, ) tracker_path = ( @@ -160,8 +167,10 @@ async def test_open_kitchen_auto_inits_tracker(self, tmp_path, track_and_release (temp_dir / ".hook_config.json").write_text("{}") steps = { - "rectify": RecipeStep(name="rectify", on_success="review_approach"), - "review_approach": RecipeStep(name="review_approach", on_success="dry_walkthrough"), + "rectify": RecipeStep(name="rectify", tool="run_skill", on_success="review_approach"), + "review_approach": RecipeStep( + name="review_approach", tool="run_skill", on_success="dry_walkthrough" + ), } ctx = _make_mock_ctx() @@ -197,8 +206,10 @@ async def test_open_kitchen_auto_init_idempotent(self, tmp_path, track_and_relea (temp_dir / ".hook_config.json").write_text("{}") steps = { - "rectify": RecipeStep(name="rectify", on_success="review_approach"), - "review_approach": RecipeStep(name="review_approach", on_success="dry_walkthrough"), + "rectify": RecipeStep(name="rectify", tool="run_skill", on_success="review_approach"), + "review_approach": RecipeStep( + name="review_approach", tool="run_skill", on_success="dry_walkthrough" + ), } ctx1 = _make_mock_ctx() @@ -251,8 +262,8 @@ async def test_open_kitchen_clears_stale_dependencies_for_existing_tracker( kitchen_id = "kitchen-clear-stale-dependencies" initial_steps = { - "rectify": RecipeStep(name="rectify", on_success="review_approach"), - "review_approach": RecipeStep(name="review_approach"), + "rectify": RecipeStep(name="rectify", tool="run_skill", on_success="review_approach"), + "review_approach": RecipeStep(name="review_approach", tool="run_skill"), } first_ctx = _make_mock_ctx() first_ctx.gate.enabled = True @@ -270,7 +281,7 @@ async def test_open_kitchen_clears_stale_dependencies_for_existing_tracker( tracker["steps"]["rectify"]["status"] = "complete" tracker_path.write_text(json.dumps(tracker)) - no_dependency_steps = {"rectify": RecipeStep(name="rectify")} + no_dependency_steps = {"rectify": RecipeStep(name="rectify", tool="run_skill")} second_ctx = _make_mock_ctx() second_ctx.gate.enabled = True second_ctx.gate_infrastructure_ready = True @@ -301,8 +312,10 @@ async def test_open_kitchen_auto_init_multi_pipeline(self, tmp_path, track_and_r (temp_dir / ".hook_config.json").write_text("{}") steps = { - "rectify": RecipeStep(name="rectify", on_success="review_approach"), - "review_approach": RecipeStep(name="review_approach", on_success="dry_walkthrough"), + "rectify": RecipeStep(name="rectify", tool="run_skill", on_success="review_approach"), + "review_approach": RecipeStep( + name="review_approach", tool="run_skill", on_success="dry_walkthrough" + ), } ctx = _make_mock_ctx() diff --git a/tests/server/test_pipeline_deps_derivation.py b/tests/server/test_pipeline_deps_derivation.py index 03ee090d6e..5e8cf3d058 100644 --- a/tests/server/test_pipeline_deps_derivation.py +++ b/tests/server/test_pipeline_deps_derivation.py @@ -14,9 +14,14 @@ def _projection( *step_names: str, edges: tuple[RecipeFlowEdge, ...] = (), + tool_by_step: dict[str, str] | None = None, ): + tool_by_step = tool_by_step or {} return _make_finalized_projection( - steps=tuple(FinalizedRecipeStep(name=name) for name in step_names), + steps=tuple( + FinalizedRecipeStep(name=name, tool=tool_by_step.get(name, "run_skill")) + for name in step_names + ), edges=edges, ) @@ -56,6 +61,139 @@ def test_review_approach_depends_on_make_plan(self): deps = _derive_phase_a_deps(projection) assert deps == {"review_approach": ["make_plan"]} + def test_run_python_predecessor_does_not_create_dependency(self): + projection = _projection( + "make_plan", + "dry_walkthrough", + edges=( + RecipeFlowEdge( + source="make_plan", + edge_type="success", + target="dry_walkthrough", + condition=None, + result_field=None, + ), + ), + tool_by_step={"make_plan": "run_python"}, + ) + + assert _derive_phase_a_deps(projection) == {} + + def test_non_skill_dependent_does_not_create_dependency(self): + projection = _projection( + "rectify", + "review_approach", + edges=( + RecipeFlowEdge( + source="rectify", + edge_type="success", + target="review_approach", + condition=None, + result_field=None, + ), + ), + tool_by_step={"review_approach": "run_cmd"}, + ) + + assert _derive_phase_a_deps(projection) == {} + + def test_ambiguous_predecessors_are_counted_before_tool_filter(self): + projection = _projection( + "rectify", + "make_plan", + "dry_walkthrough", + edges=( + RecipeFlowEdge( + source="rectify", + edge_type="success", + target="dry_walkthrough", + condition=None, + result_field=None, + ), + RecipeFlowEdge( + source="rectify", + edge_type="success", + target="make_plan", + condition=None, + result_field=None, + ), + RecipeFlowEdge( + source="make_plan", + edge_type="success", + target="dry_walkthrough", + condition=None, + result_field=None, + ), + ), + tool_by_step={"make_plan": "run_python"}, + ) + + assert _derive_phase_a_deps(projection) == {} + + def test_run_skill_predecessor_creates_dependency(self): + projection = _projection( + "rectify", + "review_approach", + edges=( + RecipeFlowEdge( + source="rectify", + edge_type="success", + target="review_approach", + condition=None, + result_field=None, + ), + ), + tool_by_step={"rectify": "run_skill", "review_approach": "run_skill"}, + ) + + assert _derive_phase_a_deps(projection) == {"review_approach": ["rectify"]} + + def test_tracker_dependencies_unions_requested_and_derived_without_mutation(self, monkeypatch): + from autoskillit.server.tools import _pipeline_deps + from autoskillit.server.tools._pipeline_deps import _tracker_dependencies + + projection = _projection( + "rectify", + "make_plan", + "review_approach", + edges=( + RecipeFlowEdge( + source="rectify", + edge_type="success", + target="review_approach", + condition=None, + result_field=None, + ), + RecipeFlowEdge( + source="review_approach", + edge_type="success", + target="make_plan", + condition=None, + result_field=None, + ), + ), + ) + derived = _derive_phase_a_deps(projection) + assert _tracker_dependencies(projection, None) == derived + + derived_by_tracker = {"review_approach": ["rectify"]} + monkeypatch.setattr( + _pipeline_deps, + "_derive_phase_a_deps", + lambda _projection: derived_by_tracker, + ) + requested = { + "review_approach": ["make_plan", "rectify"], + "custom": ["verify"], + } + + merged = _tracker_dependencies(projection, requested) + + assert set(merged["review_approach"]) == {"make_plan", "rectify"} + assert len(merged["review_approach"]) == 2 + assert merged["custom"] == ["verify"] + assert derived_by_tracker == {"review_approach": ["rectify"]} + def test_no_curated_targets_present_returns_empty(self): projection = _projection( "build", diff --git a/tests/server/test_record_pipeline_step.py b/tests/server/test_record_pipeline_step.py index c75c558a08..5ab875ceb3 100644 --- a/tests/server/test_record_pipeline_step.py +++ b/tests/server/test_record_pipeline_step.py @@ -6,6 +6,7 @@ import pytest +from autoskillit.core import FinalizedRecipeStep from autoskillit.server.tools.tools_pipeline_tracker import record_pipeline_step from tests.server._pipeline_test_helpers import _grant_success_credit @@ -18,9 +19,9 @@ def _setup(self, tool_ctx_kitchen_open, monkeypatch, tmp_path): monkeypatch.chdir(tmp_path) tool_ctx_kitchen_open.project_dir = tmp_path tool_ctx_kitchen_open.active_recipe_steps = { - "review": {}, - "implement": {}, - "verify": {}, + "review": FinalizedRecipeStep(name="review", tool="run_skill"), + "implement": FinalizedRecipeStep(name="implement", tool="run_skill"), + "verify": FinalizedRecipeStep(name="verify", tool="run_skill"), } self.ctx = tool_ctx_kitchen_open self.tmp_path = tmp_path diff --git a/tests/server/test_record_pipeline_step_admission.py b/tests/server/test_record_pipeline_step_admission.py new file mode 100644 index 0000000000..43f550d746 --- /dev/null +++ b/tests/server/test_record_pipeline_step_admission.py @@ -0,0 +1,488 @@ +"""Admission and credit flow for manual and kitchen pipeline trackers.""" + +from __future__ import annotations + +import json +import re +from pathlib import Path + +import pytest + +from autoskillit.core import FinalizedRecipeStep, RecipeFlowEdge +from autoskillit.server.tools.tools_pipeline_tracker import record_pipeline_step +from tests.server._helpers import _make_finalized_projection +from tests.server._pipeline_test_helpers import _grant_success_credit, _setup_project + +pytestmark = [pytest.mark.layer("server"), pytest.mark.medium] + + +def _tracker_path(tmp_path: Path, pipeline_id: str) -> Path: + return tmp_path / ".autoskillit" / "temp" / "pipeline_tracker" / f"{pipeline_id}.json" + + +def _install_projection( + tool_ctx, + steps: tuple[FinalizedRecipeStep, ...], + edges: tuple[RecipeFlowEdge, ...] = (), +) -> None: + steps = tuple(steps) + edges = tuple(edges) or tuple( + RecipeFlowEdge( + source=source.name, + edge_type="success", + target=target.name, + condition=None, + result_field=None, + ) + for source, target in zip(steps, steps[1:]) + ) + projection = _make_finalized_projection(steps=steps, edges=edges) + tool_ctx.active_recipe_projection = projection + tool_ctx.active_recipe_steps = {step.name: step for step in projection.ordered_steps} + + +def _install_implementation_projection(tool_ctx, tmp_path: Path) -> None: + temp_dir = tmp_path / ".autoskillit" / "temp" + temp_dir.mkdir(parents=True, exist_ok=True) + result = tool_ctx.recipes.load_and_validate( + "implementation", + project_dir=tmp_path, + temp_dir=temp_dir, + include_finalized_projection=True, + ) + projection = result["_finalized_projection"] + tool_ctx.active_recipe_projection = projection + tool_ctx.active_recipe_steps = {step.name: step for step in projection.ordered_steps} + assert projection.for_step("plan").tool == "run_skill" + assert projection.for_step("clone").tool == "bootstrap_clone" + assert projection.for_step("salvage_plan").tool == "run_python" + assert projection.for_step("claim_and_resolve") is None + assert projection.for_step("create_and_publish") is not None + + +def _install_plan_and_verify(tool_ctx) -> None: + _install_projection( + tool_ctx, + ( + FinalizedRecipeStep(name="plan", tool="run_skill"), + FinalizedRecipeStep(name="verify", tool="run_skill"), + ), + ) + + +def _release_manual_tracker_leases(tool_ctx) -> None: + from autoskillit.server._tracker_authority import _release_context_tracker + + for key in tuple(tool_ctx.tracker_leases): + if key.owner_kind == "manual": + _release_context_tracker(tool_ctx, key) + + +def _dependency_gate(tool_ctx, pipeline_id: str, step_name: str): + from autoskillit.server._tracker_authority import ( + _release_context_tracker, + _select_tracker_authority, + ) + from autoskillit.server.tools.tools_execution import _check_pipeline_deps + + target, authority, key, _lease = _select_tracker_authority( + tool_ctx, + pipeline_id, + expected=True, + ) + try: + return target, _check_pipeline_deps(step_name, authority) + finally: + if key is not None: + _release_context_tracker(tool_ctx, key) + + +@pytest.fixture +def pipeline_ctx(tool_ctx_kitchen_open, monkeypatch, tmp_path: Path): + monkeypatch.chdir(tmp_path) + tool_ctx_kitchen_open.project_dir = tmp_path + _install_plan_and_verify(tool_ctx_kitchen_open) + try: + yield tool_ctx_kitchen_open + finally: + _release_manual_tracker_leases(tool_ctx_kitchen_open) + + +@pytest.mark.anyio +async def test_manual_init_rejects_clone_edge_then_accepts_bare_init( + pipeline_ctx, + tmp_path: Path, +) -> None: + _install_implementation_projection(pipeline_ctx, tmp_path) + + result = json.loads( + await record_pipeline_step( + pipeline_id="p", + op="init", + dependencies={"plan": ["clone"]}, + ) + ) + + assert result["success"] is False + assert result["stage"] == "preflight:pipeline_tracker_dependencies" + assert result["retriable"] is False + assert "plan -> clone" in result["error"] + assert "bootstrap_clone" in result["error"] + assert "Recovery:" in result["error"] + assert not _tracker_path(tmp_path, "p").exists() + assert pipeline_ctx.tracker_leases == {} + + initialized = json.loads(await record_pipeline_step(pipeline_id="p", op="init")) + assert initialized["success"] is True + + +@pytest.mark.anyio +async def test_manual_init_rejects_retained_issue_dependency_mapping( + pipeline_ctx, + tmp_path: Path, +) -> None: + _install_implementation_projection(pipeline_ctx, tmp_path) + dependencies = { + "plan": ["clone", "claim_and_resolve", "create_and_publish"], + "salvage_plan": ["plan"], + "bind_plan_set": ["plan", "salvage_plan"], + "check_replan_iteration": ["bind_plan_set"], + "verify": ["bind_plan_set"], + "create_and_publish": ["claim_and_resolve"], + "claim_and_resolve": ["clone"], + } + + result = json.loads( + await record_pipeline_step( + pipeline_id="issue-mapping", + op="init", + dependencies=dependencies, + ) + ) + + assert result["success"] is False + assert result["stage"] == "preflight:pipeline_tracker_dependencies" + assert "plan -> clone" in result["error"] + assert "salvage_plan -> plan" in result["error"] + assert not _tracker_path(tmp_path, "issue-mapping").exists() + assert pipeline_ctx.tracker_leases == {} + + +@pytest.mark.anyio +async def test_manual_init_bounds_dependency_diagnostics( + pipeline_ctx, + tmp_path: Path, +) -> None: + pipeline_id = "diagnostic-cap" + _install_projection( + pipeline_ctx, + (FinalizedRecipeStep(name="plan", tool="run_skill"),), + ) + dependencies = {"plan": [f"missing-{index:02d}" for index in range(11)]} + + result = json.loads( + await record_pipeline_step( + pipeline_id=pipeline_id, + op="init", + dependencies=dependencies, + ) + ) + + assert result["success"] is False + assert result["stage"] == "preflight:pipeline_tracker_dependencies" + error = result["error"] + assert sum(f"plan -> missing-{index:02d}" in error for index in range(11)) == 10 + assert re.search(r"\b1\b(?=[^\n]*(?:additional|remaining|more|omitted))", error, re.I) + assert not _tracker_path(tmp_path, pipeline_id).exists() + assert pipeline_ctx.tracker_leases == {} + + +@pytest.mark.anyio +@pytest.mark.parametrize( + ("steps", "dependencies", "diagnostic"), + [ + ( + (FinalizedRecipeStep(name="plan", tool="run_skill"),), + {"plan": ["ghost"]}, + "ghost", + ), + ( + ( + FinalizedRecipeStep(name="clone", tool="bootstrap_clone"), + FinalizedRecipeStep(name="plan", tool="run_skill"), + ), + {"clone": ["plan"]}, + "bootstrap_clone", + ), + ( + (FinalizedRecipeStep(name="plan", tool="run_skill"),), + {"plan": ["plan"]}, + "self", + ), + ( + ( + FinalizedRecipeStep(name="plan", tool="run_skill"), + FinalizedRecipeStep(name="verify", tool="run_skill"), + ), + {"plan": ["verify"], "verify": ["plan"]}, + "cycle", + ), + ], + ids=["unknown-prerequisite", "non-skill-dependent", "self-edge", "cycle"], +) +async def test_synthetic_invalid_edges_are_rejected_without_a_tracker( + pipeline_ctx, + tmp_path: Path, + steps, + dependencies, + diagnostic: str, +) -> None: + pipeline_id = f"synthetic-{diagnostic}" + _install_projection(pipeline_ctx, steps) + + result = json.loads( + await record_pipeline_step( + pipeline_id=pipeline_id, + op="init", + dependencies=dependencies, + ) + ) + + assert result["success"] is False + assert result["stage"] == "preflight:pipeline_tracker_dependencies" + assert diagnostic in result["error"].casefold() + assert not _tracker_path(tmp_path, pipeline_id).exists() + assert pipeline_ctx.tracker_leases == {} + + +@pytest.mark.anyio +async def test_manual_init_rejects_empty_active_step_inventory( + pipeline_ctx, tmp_path: Path +) -> None: + pipeline_ctx.active_recipe_steps = {} + + result = json.loads(await record_pipeline_step(pipeline_id="empty-inventory", op="init")) + + assert result["success"] is False + assert "no active recipe steps" in result["error"] + assert result["stage"] == "preflight:pipeline_tracker" + assert not _tracker_path(tmp_path, "empty-inventory").exists() + assert pipeline_ctx.tracker_leases == {} + + +@pytest.mark.anyio +async def test_manual_init_is_create_only(pipeline_ctx, tmp_path: Path) -> None: + pipeline_id = "duplicate-init" + try: + first = json.loads(await record_pipeline_step(pipeline_id=pipeline_id, op="init")) + assert first["success"] is True + second = json.loads(await record_pipeline_step(pipeline_id=pipeline_id, op="init")) + + assert second["success"] is False + assert "already been initialized" in second["error"] + assert second.get("stage") + assert _tracker_path(tmp_path, pipeline_id).exists() + finally: + _release_manual_tracker_leases(pipeline_ctx) + + +@pytest.mark.anyio +async def test_manual_init_derivation_failure_releases_lease_and_writes_no_file( + pipeline_ctx, + tmp_path: Path, +) -> None: + pipeline_id = "broken-projection" + pipeline_ctx.active_recipe_projection = object() + + result = json.loads(await record_pipeline_step(pipeline_id=pipeline_id, op="init")) + + assert result["success"] is False + assert result["stage"] == "pipeline_tracker:init" + assert not _tracker_path(tmp_path, pipeline_id).exists() + assert pipeline_ctx.tracker_leases == {} + + +@pytest.mark.anyio +async def test_manual_edge_is_credited_before_dependency_gate_opens( + pipeline_ctx, + tmp_path: Path, +) -> None: + pipeline_id = "manual-credit" + try: + initialized = json.loads( + await record_pipeline_step( + pipeline_id=pipeline_id, + op="init", + dependencies={"verify": ["plan"]}, + ) + ) + assert initialized["success"] is True + + target, denial = _dependency_gate(pipeline_ctx, pipeline_id, "verify") + assert target is not None + assert denial is not None + assert "DEPENDENCY UNMET" in denial + + _grant_success_credit(pipeline_ctx, tmp_path, "plan", pipeline_id=pipeline_id) + completed = json.loads( + await record_pipeline_step( + pipeline_id=pipeline_id, + op="complete", + step_name="plan", + ) + ) + assert completed["success"] is True + tracker = json.loads(_tracker_path(tmp_path, pipeline_id).read_text()) + assert tracker["steps"]["plan"]["status"] == "complete" + + refreshed_target, admitted = _dependency_gate(pipeline_ctx, pipeline_id, "verify") + assert refreshed_target is not None + assert admitted is None + finally: + _release_manual_tracker_leases(pipeline_ctx) + + +@pytest.mark.anyio +async def test_manual_init_unions_derived_and_requested_dependencies(pipeline_ctx) -> None: + _install_projection( + pipeline_ctx, + ( + FinalizedRecipeStep(name="plan", tool="run_skill"), + FinalizedRecipeStep(name="rectify", tool="run_skill"), + FinalizedRecipeStep(name="review_approach", tool="run_skill"), + ), + edges=( + RecipeFlowEdge( + source="plan", + edge_type="success", + target="rectify", + condition=None, + result_field=None, + ), + RecipeFlowEdge( + source="rectify", + edge_type="success", + target="review_approach", + condition=None, + result_field=None, + ), + ), + ) + + derived = json.loads( + await record_pipeline_step(pipeline_id="dependency-union-derived", op="init") + ) + requested = json.loads( + await record_pipeline_step( + pipeline_id="dependency-union-requested", + op="init", + dependencies={"review_approach": ["plan"]}, + ) + ) + + assert derived["success"] is True + assert derived["dependencies"] == {"review_approach": ["rectify"]} + assert requested["success"] is True + assert requested["dependencies"] == {"review_approach": ["rectify", "plan"]} + assert requested["dependency_count"] == len(requested["dependencies"]) + + +@pytest.mark.anyio +async def test_kitchen_auto_init_tracker_can_receive_completion_credit( + tool_ctx_kitchen_open, + monkeypatch, + tmp_path: Path, +) -> None: + from autoskillit.server._tracker_authority import ( + _completion_tracker_binding, + _release_kitchen_tracker_authority, + ) + from autoskillit.server.tools.tools_kitchen._open_kitchen._tracker_auto_init import ( + _auto_init_pipeline_tracker, + ) + from tests.conftest import bind_test_kitchen_identity + + monkeypatch.chdir(tmp_path) + tool_ctx_kitchen_open.project_dir = tmp_path + bind_test_kitchen_identity(tool_ctx_kitchen_open, "kitchen-admission-credit") + _setup_project(tmp_path, tool_ctx_kitchen_open) + kitchen_id = tool_ctx_kitchen_open.kitchen_id + try: + assert _auto_init_pipeline_tracker(tool_ctx_kitchen_open) is None + target = tool_ctx_kitchen_open.kitchen_tracker_key.target + tracker_path = _tracker_path(tmp_path, kitchen_id) + tracker = json.loads(tracker_path.read_text()) + incarnation_id = tracker.get("tracker_incarnation_id") + + assert isinstance(incarnation_id, str) and incarnation_id + assert tracker["dependencies"] == {"review_approach": ["rectify"]} + kitchen_target, denial = _dependency_gate(tool_ctx_kitchen_open, "", "review_approach") + assert kitchen_target is not None + assert kitchen_target.path == target.path + assert denial is not None + assert "DEPENDENCY UNMET" in denial + + binding = _completion_tracker_binding( + tool_ctx_kitchen_open, + "", + tracker_target=target, + ) + assert len(binding) == 4 + assert binding[-1] == incarnation_id + + _grant_success_credit(tool_ctx_kitchen_open, tmp_path, "rectify", pipeline_id=kitchen_id) + completed = json.loads( + await record_pipeline_step( + pipeline_id=kitchen_id, + op="complete", + step_name="rectify", + ) + ) + assert completed["success"] is True + tracker = json.loads(tracker_path.read_text()) + assert tracker["steps"]["rectify"]["status"] == "complete" + + refreshed_target, admitted = _dependency_gate( + tool_ctx_kitchen_open, + "", + "review_approach", + ) + assert refreshed_target is not None + assert refreshed_target.path == target.path + assert admitted is None + + assert _auto_init_pipeline_tracker(tool_ctx_kitchen_open) is None + reinitialized = json.loads(tracker_path.read_text()) + assert reinitialized["tracker_incarnation_id"] == incarnation_id + assert reinitialized["steps"]["rectify"]["status"] == "complete" + finally: + if tool_ctx_kitchen_open.kitchen_tracker_key is not None: + _release_kitchen_tracker_authority( + tool_ctx_kitchen_open, + unregister=False, + retire=False, + ) + + +def test_record_pipeline_step_docstring_pins_init_admission_guidance() -> None: + docstring = (record_pipeline_step.__doc__ or "").casefold() + docstring = re.sub(r"\s+", " ", docstring) + + assert "provides the dependency graph" not in docstring + assert "idempotent" not in docstring + assert "create-only" in docstring + assert "run_skill" in docstring + assert "optional" in docstring + _, dependency_guidance = docstring.split("preflight:pipeline_tracker_dependencies", 1) + dependency_guidance, initialization_guidance = dependency_guidance.split( + "pipeline_tracker:init", 1 + ) + initialization_guidance = initialization_guidance.split('**op="status"**', 1)[0] + assert dependency_guidance + assert any(word in dependency_guidance for word in ("revis", "omit")) + assert "supplied" in dependency_guidance + assert any(word in dependency_guidance for word in ("graph", "mapping", "dependencies")) + assert initialization_guidance + assert "diagnos" in initialization_guidance + assert "initialization failure" in initialization_guidance + assert "before retrying" in initialization_guidance diff --git a/tests/server/test_record_pipeline_step_complete.py b/tests/server/test_record_pipeline_step_complete.py index 91d014343b..1fb017c4a0 100644 --- a/tests/server/test_record_pipeline_step_complete.py +++ b/tests/server/test_record_pipeline_step_complete.py @@ -14,6 +14,50 @@ class TestRecordPipelineStepComplete: + @pytest.mark.anyio + async def test_complete_refuses_linked_and_unlinked_steps_without_credit( + self, tool_ctx_kitchen_open, tmp_path + ): + from autoskillit.recipe.schema import RecipeStep + from tests.server._pipeline_test_helpers import _setup_project + + _setup_project( + tmp_path, + tool_ctx_kitchen_open, + active_recipe_steps={ + "plan": RecipeStep(name="plan", tool="run_skill", on_success="verify"), + "verify": RecipeStep(name="verify", tool="run_skill", on_success="unrelated"), + "unrelated": RecipeStep(name="unrelated", tool="run_skill"), + }, + ) + pipeline_id = "linked-steps" + initialized = json.loads( + await record_pipeline_step( + pipeline_id=pipeline_id, + op="init", + dependencies={"verify": ["plan"]}, + ) + ) + assert initialized["success"] is True + + tracker_path = ( + tmp_path / ".autoskillit" / "temp" / "pipeline_tracker" / f"{pipeline_id}.json" + ) + for step_name in ("plan", "unrelated"): + result = json.loads( + await record_pipeline_step( + pipeline_id=pipeline_id, + op="complete", + step_name=step_name, + ) + ) + + assert result["success"] is False + assert result["stage"] == "preflight:pipeline_tracker_credit" + assert result["retriable"] is False + tracker = json.loads(tracker_path.read_text()) + assert tracker["steps"][step_name]["status"] == "pending" + @pytest.mark.anyio async def test_marks_step_complete(self, tool_ctx_kitchen_open, tmp_path): tool_ctx_kitchen_open.project_dir = tmp_path diff --git a/tests/server/test_tools_status_kitchen.py b/tests/server/test_tools_status_kitchen.py index 9b5418d3e0..cf269773d5 100644 --- a/tests/server/test_tools_status_kitchen.py +++ b/tests/server/test_tools_status_kitchen.py @@ -63,8 +63,8 @@ def _failed_session_json() -> str: def _dependency_recipe_steps() -> dict[str, RecipeStep]: return { - "rectify": RecipeStep(name="rectify", on_success="review_approach"), - "review_approach": RecipeStep(name="review_approach"), + "rectify": RecipeStep(name="rectify", tool="run_skill", on_success="review_approach"), + "review_approach": RecipeStep(name="review_approach", tool="run_skill"), }