diff --git a/src/blackcell/adapters/execution/worktree.py b/src/blackcell/adapters/execution/worktree.py index 63f0b7d..fa9778b 100644 --- a/src/blackcell/adapters/execution/worktree.py +++ b/src/blackcell/adapters/execution/worktree.py @@ -563,6 +563,40 @@ def validate_base_commit(self, repository_root: Path, base_commit: str) -> None: if commit.return_code != 0: raise WorktreeLifecycleError(WorktreeFailureCode.BASE_COMMIT_NOT_FOUND) + def changed_paths_between( + self, + repository_root: Path, + *, + base_commit: str, + head_commit: str, + ) -> tuple[str, ...]: + """Return normalized cumulative path effects for one admitted commit range.""" + + root = _canonical_repository_root(repository_root) + self.validate_base_commit(root, base_commit) + self.validate_base_commit(root, head_commit) + ancestor = self._git_at( + root, + ("merge-base", "--is-ancestor", base_commit, head_commit), + ) + if ancestor.return_code != 0: + raise WorktreeLifecycleError(WorktreeFailureCode.WORKTREE_CONFLICT) + changed = self._require_success( + self._git_at( + root, + ( + "diff", + "--name-only", + "--no-renames", + "-z", + base_commit, + head_commit, + "--", + ), + ) + ) + return self._path_output(changed) + def retain_plan_base_commit( self, repository_root: Path, diff --git a/src/blackcell/bootstrap/execution_plan.py b/src/blackcell/bootstrap/execution_plan.py index f805564..96f7062 100644 --- a/src/blackcell/bootstrap/execution_plan.py +++ b/src/blackcell/bootstrap/execution_plan.py @@ -3,9 +3,11 @@ from __future__ import annotations from collections.abc import Callable, Mapping -from dataclasses import dataclass, field +from dataclasses import dataclass, field, replace +from math import ceil, isfinite from pathlib import Path -from typing import Protocol +from time import monotonic +from typing import Literal, Protocol from blackcell.adapters.execution.evidence import ExecutionEvidenceCollector, ExecutionEvidenceError from blackcell.adapters.execution.text_changes import ( @@ -49,8 +51,22 @@ change_proposal_payload, change_provider_result_payload, ) +from blackcell.orchestration.execution_artifacts import ( + ACCEPTANCE_COMMAND_MEDIA_TYPE, + ACCEPTANCE_RESULT_MEDIA_TYPE, + CONTEXT_MEDIA_TYPE, + EFFECT_MEDIA_TYPE, + OUTCOME_MEDIA_TYPE, + PROPOSAL_MEDIA_TYPE, + PROVIDER_MEDIA_TYPE, + CheckArtifacts, + ExecutionArtifactLink, + NodeOutcomeManifest, + node_outcome_payload, +) from blackcell.orchestration.execution_plan import ( AttemptEvidence, + ExecutionAuthority, FailureClass, GoalSpec, Plan, @@ -143,6 +159,15 @@ def __init__(self, code: str = "execution-execution-failed") -> None: super().__init__(code) +@dataclass(slots=True) +class _AttemptArtifacts: + context: ExecutionArtifactLink | None = None + proposal: ExecutionArtifactLink | None = None + provider: ExecutionArtifactLink | None = None + effect: ExecutionArtifactLink | None = None + checks: list[CheckArtifacts] = field(default_factory=list) + + @dataclass(frozen=True, slots=True) class ProductionAttemptExecutor: """Execute admitted tasks through real worktree, proposal, effect, and sandbox ports.""" @@ -157,10 +182,17 @@ class ProductionAttemptExecutor: evidence: ExecutionEvidenceCollector | None = field(default=None, repr=False) changes: TextChangeExecutor | None = field(default=None, repr=False) cancel_requested: Callable[[str], bool] | None = field(default=None, repr=False) + authority_for_run: Callable[[str], ExecutionAuthority] | None = field( + default=None, + repr=False, + ) + clock: Callable[[], float] = field(default=monotonic, repr=False, compare=False) def __post_init__(self) -> None: if not isinstance(self.repository_root, Path) or not isinstance(self.isolation_root, Path): raise ValueError("invalid execution execution roots") + if not callable(self.clock): + raise ValueError("invalid execution clock") object.__setattr__(self, "repository_root", self.repository_root.resolve(strict=True)) object.__setattr__(self, "isolation_root", self.isolation_root.resolve(strict=True)) if self.evidence is None: @@ -193,6 +225,19 @@ def execute( ) if not policy_decision.allowed or policy_decision.action_digest != expected_action.digest: raise ExecutionError("execution-policy-denied") + authority = self._authority(run_id) + authority_budget = _subtract_budget( + authority.budget, + authority.consumed_budget, + ) + latency_authority_ms = min( + remaining_budget.max_latency_ms, + authority_budget.max_latency_ms, + ) + max_changed_paths = min( + self.policy.max_changed_paths, + authority.max_changed_paths, + ) self._require_not_canceled(run_id) lease = WorktreeLeaseIdentity( run_id=run_id, @@ -207,9 +252,10 @@ def execute( isolation_root=self.isolation_root, base_commit=base_commit, allowed_paths=task.allowed_paths, - max_changed_paths=self.policy.max_changed_paths, + max_changed_paths=max_changed_paths, ) artifacts: list[str] = [] + stages = _AttemptArtifacts() semantic: list[dict[str, JsonInput]] = [] input_tokens: int | None = 0 output_tokens: int | None = 0 @@ -235,8 +281,24 @@ def execute( media_type="application/vnd.blackcell.worktree-inspection+json", ).digest ) + prior_changed_paths = self.worktrees.changed_paths_between( + self.repository_root, + base_commit=goal.base_commit, + head_commit=base_commit, + ) + if len(prior_changed_paths) > max_changed_paths: + raise ExecutionError("execution-cumulative-path-limit-exceeded") commit_effects: tuple[WorktreeCommitEffect, ...] = () if task.allowed_paths: + call_budget = _intersect_budget( + _intersect_budget(self.policy.provider_budget, remaining_budget), + authority_budget, + ) + if _authority_usage_incomplete(authority) or _provider_budget_exhausted( + call_budget, + total_budget=authority.budget, + ): + raise ExecutionError("execution-cumulative-budget-exhausted") self._require_not_canceled(run_id) context = self._evidence().collect( spec, @@ -249,12 +311,12 @@ def execute( ) context_ref = self._store_json( change_context_payload(context), - media_type="application/vnd.blackcell.context+json", + media_type=CONTEXT_MEDIA_TYPE, expected_digest=context.digest, ) + stages.context = context_ref artifacts.append(context_ref.digest) semantic.append({"kind": "context", "digest": context_ref.digest}) - call_budget = _intersect_budget(self.policy.provider_budget, remaining_budget) input_tokens = None output_tokens = None cost_microusd = None @@ -282,15 +344,20 @@ def execute( cost_microusd = provider_result.cost_microusd if _usage_overdraws_budget(provider_result, call_budget): raise ExecutionError("execution-cumulative-budget-exhausted") + proposed_paths = {item.path for item in provider_result.proposal.operations} + if len(set(prior_changed_paths) | proposed_paths) > max_changed_paths: + raise ExecutionError("execution-cumulative-path-limit-exceeded") proposal_ref = self._store_json( change_proposal_payload(provider_result.proposal), - media_type="application/vnd.blackcell.proposal+json", + media_type=PROPOSAL_MEDIA_TYPE, expected_digest=provider_result.proposal.digest, ) provider_ref = self._store_json( change_provider_result_payload(provider_result), - media_type="application/vnd.blackcell.provider+json", + media_type=PROVIDER_MEDIA_TYPE, ) + stages.proposal = proposal_ref + stages.provider = provider_ref artifacts.extend((proposal_ref.digest, provider_ref.digest)) semantic.append({"kind": "proposal", "digest": proposal_ref.digest}) effect = self._changes().execute( @@ -305,9 +372,10 @@ def execute( ) effect_ref = self._store_json( text_change_result_payload(effect), - media_type="application/vnd.blackcell.effect+json", + media_type=EFFECT_MEDIA_TYPE, expected_digest=effect.result_digest, ) + stages.effect = effect_ref artifacts.append(effect_ref.digest) commit_effects = tuple( WorktreeCommitEffect(item.path, item.after_digest) for item in effect.effects @@ -315,37 +383,86 @@ def execute( self._require_not_canceled(run_id) committed = self.worktrees.commit_changes(spec, effects=commit_effects) + cumulative_changed_paths = self.worktrees.changed_paths_between( + self.repository_root, + base_commit=goal.base_commit, + head_commit=committed.head_commit, + ) + if len(cumulative_changed_paths) > max_changed_paths: + raise ExecutionError("execution-cumulative-path-limit-exceeded") check_evidence: list[dict[str, JsonInput]] = [] failed: list[AcceptanceResult] = [] for check in task.checks: self._require_not_canceled(run_id) + available_latency_ms = latency_authority_ms - latency_ms + if available_latency_ms <= 0: + raise ExecutionError("execution-cumulative-budget-exhausted") command = AcceptanceCommand( check_id=check.check_id, argv=check.argv, expected_exit_code=check.expected_exit_code, - timeout_seconds=self.policy.check_timeout_seconds, + timeout_seconds=min( + self.policy.check_timeout_seconds, + authority.check_timeout_seconds, + available_latency_ms / 1_000, + ), stdout_limit_bytes=self.policy.stdout_limit_bytes, stderr_limit_bytes=self.policy.stderr_limit_bytes, ) - result = self.acceptance.run( - command, - spec, - cancel_requested=lambda: self._is_canceled(run_id), - ) + check_started = _clock_sample(self.clock) + result: AcceptanceResult | None = None + runner_error: Exception | None = None + try: + result = self.acceptance.run( + command, + spec, + cancel_requested=lambda: self._is_canceled(run_id), + ) + except Exception as error: + runner_error = error + try: + elapsed_ms = _elapsed_milliseconds( + check_started, + _clock_sample(self.clock), + ) + except ExecutionError: + latency_ms = latency_authority_ms + raise + latency_ms += elapsed_ms + if latency_ms > latency_authority_ms: + raise ExecutionError("execution-cumulative-budget-exhausted") + if runner_error is not None: + if isinstance(runner_error, KernelError | OSError): + raise ExecutionError("execution-acceptance-runner-failed") from runner_error + raise runner_error + if result is None: + raise ExecutionError("invalid-execution-acceptance-result") self._require_not_canceled(run_id) _validate_check(command, spec, result) command_ref = self._store_json( acceptance_command_payload(command), - media_type="application/vnd.blackcell.check-command+json", + media_type=ACCEPTANCE_COMMAND_MEDIA_TYPE, expected_digest=command.digest, ) result_ref = self._store_json( acceptance_result_payload(result), - media_type="application/vnd.blackcell.check-result+json", + media_type=ACCEPTANCE_RESULT_MEDIA_TYPE, expected_digest=result.digest, ) - stdout_ref = self.artifacts.put_bytes(result.stdout.captured) - stderr_ref = self.artifacts.put_bytes(result.stderr.captured) + stdout_ref = self._store_bytes(result.stdout.captured) + stderr_ref = self._store_bytes(result.stderr.captured) + stages.checks.append( + CheckArtifacts( + check_id=result.check_id, + command_digest=command.digest, + result_digest=result.digest, + passed=result.passed, + command=command_ref, + result=result_ref, + stdout=stdout_ref, + stderr=stderr_ref, + ) + ) artifacts.extend( (command_ref.digest, result_ref.digest, stdout_ref.digest, stderr_ref.digest) ) @@ -373,6 +490,7 @@ def execute( workspace_id=workspace_id, policy_decision=policy_decision, artifacts=artifacts, + stages=stages, semantic=semantic, input_tokens=input_tokens, output_tokens=output_tokens, @@ -391,6 +509,7 @@ def execute( workspace_id=workspace_id, policy_decision=policy_decision, artifacts=artifacts, + stages=stages, semantic=semantic, check_evidence=check_evidence, workspace_clean=True, @@ -425,6 +544,7 @@ def execute( workspace_id=workspace_id, policy_decision=policy_decision, artifacts=artifacts, + stages=stages, semantic=semantic, check_evidence=check_evidence, workspace_clean=True, @@ -450,6 +570,7 @@ def execute( workspace_id=workspace_id, policy_decision=policy_decision, artifacts=artifacts, + stages=stages, semantic=semantic, input_tokens=input_tokens, output_tokens=output_tokens, @@ -469,6 +590,7 @@ def _failure( workspace_id: str, policy_decision: PolicyDecision, artifacts: list[str], + stages: _AttemptArtifacts, semantic: list[dict[str, JsonInput]], input_tokens: int | None, output_tokens: int | None, @@ -490,6 +612,7 @@ def _failure( workspace_id=workspace_id, policy_decision=policy_decision, artifacts=artifacts, + stages=stages, semantic=semantic, check_evidence=[], workspace_clean=inspection.clean and inspection.path_policy_compliant, @@ -515,6 +638,7 @@ def _finish( workspace_id: str, policy_decision: PolicyDecision, artifacts: list[str], + stages: _AttemptArtifacts, semantic: list[dict[str, JsonInput]], check_evidence: list[dict[str, JsonInput]], workspace_clean: bool, @@ -528,6 +652,31 @@ def _finish( latency_ms: int, cost_microusd: int | None, ) -> AttemptEvidence: + status: Literal["succeeded", "failed"] = "succeeded" if failure_class is None else "failed" + outcome = NodeOutcomeManifest( + run_id=run_id, + node_id=task.task_id, + attempt=attempt, + fencing_token=spec.lease.fencing_token, + lease_digest=spec.lease.digest, + worktree_spec_digest=spec.digest, + base_commit=spec.base_commit, + head_commit=head_commit, + repository_write=bool(task.allowed_paths), + status=status, + failure_code=None if failure_class is None else failure_class.value, + context_artifact=stages.context, + proposal_artifact=stages.proposal, + provider_artifact=stages.provider, + effect_artifact=stages.effect, + checks=tuple(stages.checks), + ) + outcome_ref = self._store_json( + node_outcome_payload(outcome), + media_type=OUTCOME_MEDIA_TYPE, + expected_digest=outcome.digest, + ) + artifacts.append(outcome_ref.digest) semantic_payload: dict[str, JsonInput] = { "schema_version": "blackcell.execution-evidence/v1", "workspace_clean": workspace_clean, @@ -596,7 +745,7 @@ def _store_json( *, media_type: str, expected_digest: str | None = None, - ) -> ArtifactRef: + ) -> ExecutionArtifactLink: reference = self.artifacts.put_bytes( canonical_json_bytes(dict(payload)), media_type=media_type, @@ -604,7 +753,10 @@ def _store_json( ) if expected_digest is not None and reference.digest != expected_digest: raise ExecutionError("execution-artifact-digest-mismatch") - return reference + return ExecutionArtifactLink.from_reference(reference) + + def _store_bytes(self, payload: bytes) -> ExecutionArtifactLink: + return ExecutionArtifactLink.from_reference(self.artifacts.put_bytes(payload)) def _evidence(self) -> ExecutionEvidenceCollector: if self.evidence is None: # pragma: no cover - established in __post_init__ @@ -619,6 +771,18 @@ def _changes(self) -> TextChangeExecutor: def _is_canceled(self, run_id: str) -> bool: return self.cancel_requested is not None and self.cancel_requested(run_id) + def _authority(self, run_id: str) -> ExecutionAuthority: + if self.authority_for_run is None: + return ExecutionAuthority( + budget=self.policy.provider_budget, + check_timeout_seconds=self.policy.check_timeout_seconds, + max_changed_paths=self.policy.max_changed_paths, + ) + authority = self.authority_for_run(run_id) + if not isinstance(authority, ExecutionAuthority): + raise ExecutionError("execution-authority-invalid") + return authority + def _require_not_canceled(self, run_id: str) -> None: if self._is_canceled(run_id): raise ExecutionError("execution-canceled") @@ -637,28 +801,44 @@ class ProductionExecution: """Application service that consumes the coordinator in the production process graph.""" coordinator: ExecutionCoordinator + authority_for_run: Callable[[str], ExecutionAuthority] | None = field( + default=None, + repr=False, + ) + goal_for_run: Callable[[str], GoalSpec] | None = field(default=None, repr=False) def process(self, request: PlanningRequest, *, actor: str) -> ExecutionRunState: """Start or safely resume one public generated-plan run.""" + self._require_goal_binding(request) + worker_budget = request.budget try: state = self.coordinator.journal.rehydrate(request.run_id) except ExecutionRuntimeError as error: if error.code != "execution-run-not-found": raise + authority = self._authority(request.run_id) + request = self._bounded_request(request, authority) + provider_budget = self._provider_budget(worker_budget, authority) + if _authority_usage_incomplete(authority) or _provider_budget_exhausted( + provider_budget, + total_budget=request.budget, + ): + raise ExecutionError("execution-cumulative-budget-exhausted") from None plan, _ = self.coordinator.compile_and_admit( request.run_id, request, actor=actor, + provider_budget=provider_budget, ) return self.coordinator.execute(request.run_id, request, plan, actor=actor) if ( state.goal_digest != request.goal.digest or state.classification is not request.classification or state.locality is not request.locality - or state.budget != request.budget ): raise ExecutionRuntimeError("execution-run-binding-mismatch") + request = replace(request, budget=state.budget) if state.status in { RunLifecycleStatus.SUCCEEDED, RunLifecycleStatus.BLOCKED, @@ -673,11 +853,26 @@ def process(self, request: PlanningRequest, *, actor: str) -> ExecutionRunState: if state.status is RunLifecycleStatus.REPLAN_REQUIRED: if plan.plan_revision >= 2: return self.coordinator.exhaust_replan_budget(request.run_id, actor=actor) + authority = self._authority(request.run_id) + provider_budget = self._provider_budget( + _intersect_budget(worker_budget, _remaining_state_budget(state)), + authority, + ) + if ( + _provider_usage_incomplete(state) + or _authority_usage_incomplete(authority) + or _provider_budget_exhausted( + provider_budget, + total_budget=authority.budget if authority is not None else state.budget, + ) + ): + return self.coordinator.exhaust_provider_budget(request.run_id, actor=actor) plan, _ = self.coordinator.compile_and_admit( request.run_id, request, actor=actor, previous=plan, + provider_budget=provider_budget, ) return self.coordinator.execute(request.run_id, request, plan, actor=actor) @@ -688,11 +883,43 @@ def run( actor: str, previous: Plan | None = None, ) -> ExecutionRunResult: + self._require_goal_binding(request) + worker_budget = request.budget + authority = self._authority(request.run_id) + durable_usage_incomplete = False + if previous is None: + request = self._bounded_request(request, authority) + durable_remainder = request.budget + else: + state = self.coordinator.journal.rehydrate(request.run_id) + if ( + state.goal_digest != request.goal.digest + or state.classification is not request.classification + or state.locality is not request.locality + ): + raise ExecutionRuntimeError("execution-run-binding-mismatch") + request = replace(request, budget=state.budget) + durable_remainder = _remaining_state_budget(state) + durable_usage_incomplete = _provider_usage_incomplete(state) + provider_budget = self._provider_budget( + _intersect_budget(worker_budget, durable_remainder), + authority, + ) + if ( + durable_usage_incomplete + or _authority_usage_incomplete(authority) + or _provider_budget_exhausted( + provider_budget, + total_budget=(authority.budget if authority is not None else request.budget), + ) + ): + raise ExecutionError("execution-cumulative-budget-exhausted") plan, planning = self.coordinator.compile_and_admit( request.run_id, request, actor=actor, previous=previous, + provider_budget=provider_budget, ) state = self.coordinator.execute( request.run_id, @@ -707,6 +934,45 @@ def run( promotion_candidate=self.coordinator.journal.promotion_candidate(request.run_id), ) + def _authority(self, run_id: str) -> ExecutionAuthority | None: + if self.authority_for_run is None: + return None + authority = self.authority_for_run(run_id) + if not isinstance(authority, ExecutionAuthority): + raise ExecutionError("execution-authority-invalid") + return authority + + def _require_goal_binding(self, request: PlanningRequest) -> None: + if self.goal_for_run is None: + return + expected = self.goal_for_run(request.run_id) + if not isinstance(expected, GoalSpec): + raise ExecutionError("execution-goal-authority-invalid") + if request.goal != expected: + raise ExecutionRuntimeError("execution-run-binding-mismatch") + + @staticmethod + def _bounded_request( + request: PlanningRequest, + authority: ExecutionAuthority | None, + ) -> PlanningRequest: + if authority is None: + return request + bounded = _intersect_budget(request.budget, authority.budget) + return request if bounded == request.budget else replace(request, budget=bounded) + + @staticmethod + def _provider_budget( + worker_budget: GatewayBudget, + authority: ExecutionAuthority | None, + ) -> GatewayBudget: + if authority is None: + return worker_budget + return _intersect_budget( + worker_budget, + _subtract_budget(authority.budget, authority.consumed_budget), + ) + def _repair_constraints( constraints: tuple[str, ...], @@ -732,6 +998,87 @@ def _intersect_budget(left: GatewayBudget, right: GatewayBudget) -> GatewayBudge ) +def _clock_sample(clock: Callable[[], float]) -> float: + try: + value = clock() + except Exception as error: + raise ExecutionError("execution-clock-invalid") from error + if isinstance(value, bool) or not isinstance(value, int | float) or not isfinite(value): + raise ExecutionError("execution-clock-invalid") + return float(value) + + +def _elapsed_milliseconds(started: float, finished: float) -> int: + if not isfinite(started) or not isfinite(finished) or finished < started: + raise ExecutionError("execution-clock-invalid") + elapsed_seconds = finished - started + if not isfinite(elapsed_seconds): + raise ExecutionError("execution-clock-invalid") + elapsed_milliseconds = elapsed_seconds * 1_000 + if not isfinite(elapsed_milliseconds): + raise ExecutionError("execution-clock-invalid") + return ceil(elapsed_milliseconds) + + +def _subtract_budget(total: GatewayBudget, consumed: GatewayBudget) -> GatewayBudget: + return GatewayBudget( + max(0, total.max_input_tokens - consumed.max_input_tokens), + max(0, total.max_output_tokens - consumed.max_output_tokens), + max(0, total.max_latency_ms - consumed.max_latency_ms), + max(0, total.max_cost_microusd - consumed.max_cost_microusd), + ) + + +def _remaining_state_budget(state: ExecutionRunState) -> GatewayBudget: + return GatewayBudget( + ( + max(0, state.budget.max_input_tokens - state.input_tokens) + if state.input_tokens_complete + else 0 + ), + ( + max(0, state.budget.max_output_tokens - state.output_tokens) + if state.output_tokens_complete + else 0 + ), + max(0, state.budget.max_latency_ms - state.latency_ms), + ( + max(0, state.budget.max_cost_microusd - state.cost_microusd) + if state.cost_microusd_complete + else 0 + ), + ) + + +def _authority_usage_incomplete(authority: ExecutionAuthority | None) -> bool: + return authority is not None and ( + not authority.input_tokens_complete + or not authority.output_tokens_complete + or not authority.cost_microusd_complete + ) + + +def _provider_usage_incomplete(state: ExecutionRunState) -> bool: + return ( + not state.input_tokens_complete + or not state.output_tokens_complete + or not state.cost_microusd_complete + ) + + +def _provider_budget_exhausted( + budget: GatewayBudget, + *, + total_budget: GatewayBudget, +) -> bool: + return ( + budget.max_input_tokens == 0 + or budget.max_output_tokens == 0 + or budget.max_latency_ms == 0 + or (total_budget.max_cost_microusd > 0 and budget.max_cost_microusd == 0) + ) + + def _usage_overdraws_budget( result: ChangeProviderResult, budget: GatewayBudget, diff --git a/src/blackcell/bootstrap/execution_process.py b/src/blackcell/bootstrap/execution_process.py index 8f2a69e..b0a3149 100644 --- a/src/blackcell/bootstrap/execution_process.py +++ b/src/blackcell/bootstrap/execution_process.py @@ -62,7 +62,9 @@ MAX_CHANGE_PROPOSAL_BYTES, ) from blackcell.orchestration.execution_plan import ( + ExecutionAuthority, ExecutionPolicyKernel, + GoalSpec, PlanningRequest, RunLifecycleStatus, planning_payload, @@ -101,6 +103,10 @@ def run_once(self) -> ExecutionWorkerCycleResult: ... class ExecutionReconciliationPort(Protocol): def next_generated_run(self) -> GeneratedRun | None: ... + def generated_execution_goal(self, run_id: str) -> GoalSpec: ... + + def generated_execution_authority(self, run_id: str) -> ExecutionAuthority: ... + def should_cancel_generated_run(self, run_id: str) -> bool: ... def reconcile_startup(self, *, principal_id: str) -> tuple[object, ...]: ... @@ -202,6 +208,7 @@ def from_config( evidence=ExecutionEvidenceCollector(boundaries.worktrees), changes=TextChangeExecutor(boundaries.worktrees), cancel_requested=runtime.should_cancel_generated_run, + authority_for_run=runtime.generated_execution_authority, ) observer = ( None if telemetry.recorder is None else ExecutionTraceObserver(telemetry.recorder) @@ -216,7 +223,9 @@ def from_config( boundaries.planner, execution_executor, ExecutionPolicyKernel(), - ) + ), + authority_for_run=runtime.generated_execution_authority, + goal_for_run=runtime.generated_execution_goal, ) except Exception: telemetry.shutdown() @@ -253,11 +262,14 @@ def _run_generated_once( generated = self.runtime.next_generated_run() if generated is None: return None - budget = GatewayBudget( - config.provider.max_input_tokens, - config.provider.max_output_tokens, - config.provider.timeout_ceiling_seconds * 1_000, - config.provider.max_cost_microusd, + budget = _intersect_budget( + GatewayBudget( + config.provider.max_input_tokens, + config.provider.max_output_tokens, + config.provider.timeout_ceiling_seconds * 1_000, + config.provider.max_cost_microusd, + ), + generated.authority.budget, ) payload_size = len(canonical_json_bytes(planning_payload(generated.goal))) request = PlanningRequest( @@ -336,6 +348,15 @@ def serve(self, *, once: bool = False) -> int: self.shutdown() +def _intersect_budget(left: GatewayBudget, right: GatewayBudget) -> GatewayBudget: + return GatewayBudget( + min(left.max_input_tokens, right.max_input_tokens), + min(left.max_output_tokens, right.max_output_tokens), + min(left.max_latency_ms, right.max_latency_ms), + min(left.max_cost_microusd, right.max_cost_microusd), + ) + + def validate_execution_worker_runtime_config( config: RuntimeProcessConfig, *, diff --git a/src/blackcell/bootstrap/runtime_service.py b/src/blackcell/bootstrap/runtime_service.py index c40241f..72a8be6 100644 --- a/src/blackcell/bootstrap/runtime_service.py +++ b/src/blackcell/bootstrap/runtime_service.py @@ -3,6 +3,7 @@ from collections.abc import Mapping from dataclasses import dataclass from datetime import datetime +from math import isfinite from pathlib import Path from typing import cast @@ -23,6 +24,7 @@ worktree_removal_payload, ) from blackcell.config import RuntimeSecurityConfig +from blackcell.gateway import GatewayBudget from blackcell.interfaces.http.contracts import ( MAX_RUN_QUERY_SCAN_EVENTS, MAX_RUNTIME_EVENT_PAGE_SIZE, @@ -68,12 +70,20 @@ utc_now, ) from blackcell.kernel._json import JsonInput, bytes_digest, json_digest, thaw_json +from blackcell.orchestration.execution_artifacts import ( + OUTCOME_MEDIA_TYPE, + NodeOutcomeManifest, + node_outcome_from_mapping, +) from blackcell.orchestration.execution_plan import ( + EXECUTION_GOAL_ADMITTED, EXECUTION_PLAN_ADMITTED, EXECUTION_TASK_VERIFIED, + ExecutionAuthority, GoalSpec, Plan, VerificationCheck, + goal_from_payload, plan_from_payload, ) from blackcell.orchestration.execution_plan import ( @@ -86,6 +96,7 @@ ExecutionRunProjection, ExecutionRunState, ExecutionRuntimeError, + ExecutionTaskState, ) from blackcell.orchestration.replay import ( ExecutionArtifactReaderPort, @@ -182,6 +193,7 @@ class GeneratedRun: run_id: str goal: GoalSpec + authority: ExecutionAuthority @dataclass(frozen=True, slots=True) @@ -535,9 +547,54 @@ def next_generated_run(self) -> GeneratedRun | None: or (kernel is not None and kernel.status in _EXECUTION_TERMINAL_STATUSES) ): continue - return GeneratedRun(run_id, _generated_goal(loaded)) + return GeneratedRun(run_id, _generated_goal(loaded), _generated_authority(loaded.plan)) return None + def generated_execution_authority(self, run_id: str) -> ExecutionAuthority: + """Re-read the immutable public-plan bounds for one generated execution attempt.""" + + loaded = self._load_run(run_id) + if loaded.plan.planning_mode != "generated": + raise RuntimeApiError(RuntimeApiFailureCode.CONFLICT) + authority = _generated_authority(loaded.plan) + kernel = loaded.kernel_state + if kernel is None: + return authority + return ExecutionAuthority( + budget=authority.budget, + check_timeout_seconds=authority.check_timeout_seconds, + max_changed_paths=authority.max_changed_paths, + consumed_budget=GatewayBudget( + ( + kernel.input_tokens + if kernel.input_tokens_complete + else authority.budget.max_input_tokens + ), + ( + kernel.output_tokens + if kernel.output_tokens_complete + else authority.budget.max_output_tokens + ), + kernel.latency_ms, + ( + kernel.cost_microusd + if kernel.cost_microusd_complete + else authority.budget.max_cost_microusd + ), + ), + input_tokens_complete=kernel.input_tokens_complete, + output_tokens_complete=kernel.output_tokens_complete, + cost_microusd_complete=kernel.cost_microusd_complete, + ) + + def generated_execution_goal(self, run_id: str) -> GoalSpec: + """Re-read the canonical goal derived from one accepted generated plan.""" + + loaded = self._load_run(run_id) + if loaded.plan.planning_mode != "generated": + raise RuntimeApiError(RuntimeApiFailureCode.CONFLICT) + return _generated_goal(loaded) + def should_cancel_node(self, spec: WorktreeExecutionSpec) -> bool: """Return true when cancellation or fencing requires an active worker to stop.""" @@ -1354,7 +1411,7 @@ def review_run_ids(self) -> tuple[str, ...]: run_ids: list[str] = [] for run_id in self._run_ids(): loaded = self._load_run(run_id) - if loaded.state.status is RunLifecycleStatus.SUCCEEDED: + if _effective_run_status(loaded) == "succeeded": run_ids.append(run_id) return tuple(run_ids) @@ -1363,7 +1420,7 @@ def review_candidate(self, run_id: str) -> ReviewCandidate: _identifier(run_id) loaded = self._load_run(run_id) - if loaded.state.status is not RunLifecycleStatus.SUCCEEDED: + if _effective_run_status(loaded) != "succeeded": raise RuntimeApiError(RuntimeApiFailureCode.CONFLICT) return self._review_candidate(loaded) @@ -1374,10 +1431,16 @@ def prepare_review_context(self, candidate: ReviewCandidate) -> ReviewContext: raise RuntimeApiError(RuntimeApiFailureCode.INVALID_REQUEST) loaded = self._load_run(candidate.run_id) if ( - loaded.state.status is not RunLifecycleStatus.SUCCEEDED + _effective_run_status(loaded) != "succeeded" or self._review_candidate(loaded) != candidate ): raise RuntimeApiError(RuntimeApiFailureCode.CONFLICT) + expectations = self._review_artifact_expectations(loaded) + review_constraints = ( + loaded.intent.constraints + if loaded.kernel_state is None + else _kernel_goal(loaded).constraints + ) return build_review_context_from_artifacts( self._artifacts, run_id=candidate.run_id, @@ -1385,21 +1448,15 @@ def prepare_review_context(self, candidate: ReviewCandidate) -> ReviewContext: intent_id=loaded.request.intent_id, plan_id=loaded.request.plan_id, objective=loaded.intent.objective, - constraints=loaded.intent.constraints, + constraints=review_constraints, base_commit=loaded.plan.base_commit, state_digest=candidate.state_digest, - nodes=_artifact_expectations(loaded), + nodes=expectations, + require_exact_constraints=loaded.kernel_state is None, ) def _review_candidate(self, loaded: _LoadedRun) -> ReviewCandidate: - terminal_index = next( - ( - index - for index, event in reversed(tuple(enumerate(loaded.events))) - if event.event_type == RUN_SUCCEEDED - ), - None, - ) + terminal_index = _review_terminal_index(loaded) if terminal_index is None: raise RuntimeApiError(RuntimeApiFailureCode.CONFLICT) terminal = loaded.events[terminal_index] @@ -1420,19 +1477,34 @@ def _review_candidate(self, loaded: _LoadedRun) -> ReviewCandidate: _plan_stream(loaded.request.plan_id), _PLAN_ACCEPTED, ) - state_digest = json_digest( - { - "project": _request_value(project_event), - "intent": _request_value(intent_event), - "plan": _request_value(plan_event), - "run": _request_value(loaded.state.queued_event), - "lifecycle": run_lifecycle_payload(terminal_state), - } - ) + state_payload: dict[str, object] = { + "project": _request_value(project_event), + "intent": _request_value(intent_event), + "plan": _request_value(plan_event), + "run": _request_value(loaded.state.queued_event), + "lifecycle": run_lifecycle_payload(terminal_state), + } + if loaded.kernel_state is not None: + terminal_kernel = ( + ProjectionRunner() + .replay( + ExecutionRunProjection(), + loaded.events[: terminal_index + 1], + ) + .state + ) + if ( + terminal_kernel is None + or terminal_kernel.status is not ExecutionRunStatus.SUCCEEDED + ): + raise RuntimeApiError(RuntimeApiFailureCode.CONFLICT) + state_payload["kernel"] = ExecutionRunProjection().dump_state(terminal_kernel) + state_digest = json_digest(cast("Mapping[str, JsonInput]", state_payload)) + expectations = self._review_artifact_expectations(loaded) artifact_report = verify_run_artifacts( self._artifacts, run_id=loaded.request.run_id, - nodes=_artifact_expectations(loaded), + nodes=expectations, ) return ReviewCandidate( run_id=loaded.request.run_id, @@ -1444,6 +1516,105 @@ def _review_candidate(self, loaded: _LoadedRun) -> ReviewCandidate: artifact_evidence_digest=artifact_report.evidence_digest, ) + def _review_artifact_expectations( + self, + loaded: _LoadedRun, + ) -> tuple[ReplayNodeExpectation, ...]: + if loaded.kernel_state is None: + return _artifact_expectations(loaded) + if self._artifacts is None: + raise RuntimeApiError(RuntimeApiFailureCode.CONFLICT) + plan = _kernel_plan(loaded) + goal = _kernel_goal(loaded) + states = {item.task_id: item for item in loaded.kernel_state.tasks} + authority = _generated_authority(loaded.plan) + expectations: list[ReplayNodeExpectation] = [] + for task in plan.tasks: + state = states.get(task.task_id) + if state is None: + raise RuntimeApiError(RuntimeApiFailureCode.CONFLICT) + outcome_digest, outcome = _kernel_outcome( + self._artifacts, + state.last_artifact_digests, + ) + if ( + state.status is not ExecutionTaskStatus.SUCCEEDED + or outcome.run_id != loaded.request.run_id + or outcome.node_id != task.task_id + or outcome.attempt != state.attempts + or outcome.status != "succeeded" + ): + raise RuntimeApiError(RuntimeApiFailureCode.CONFLICT) + base_commit = _kernel_task_base_commit( + plan, + states, + task.task_id, + ) + constraints = goal.constraints + max_changed_paths = 0 + provider_context_digest = None + if task.allowed_paths: + context = outcome.context_artifact + if context is None: + raise RuntimeApiError(RuntimeApiFailureCode.CONFLICT) + raw_context = _artifact_mapping(self._artifacts, context.digest) + constraints = _kernel_context_constraints( + raw_context, + accepted=goal.constraints, + ) + max_changed_paths = _artifact_integer(raw_context, "max_changed_paths") + if max_changed_paths > authority.max_changed_paths: + raise RuntimeApiError(RuntimeApiFailureCode.CONFLICT) + provider_context_digest = context.digest + recorded_checks = {item.check_id: item for item in outcome.checks} + checks: list[ReplayCheckExpectation] = [] + for check in task.checks: + recorded = recorded_checks.get(check.check_id) + if recorded is None: + raise RuntimeApiError(RuntimeApiFailureCode.CONFLICT) + command = _artifact_mapping(self._artifacts, recorded.command.digest) + timeout_seconds = _artifact_number(command, "timeout_seconds") + if timeout_seconds > authority.check_timeout_seconds: + raise RuntimeApiError(RuntimeApiFailureCode.CONFLICT) + checks.append( + ReplayCheckExpectation( + check.check_id, + check.argv, + check.expected_exit_code, + timeout_seconds, + ) + ) + expectations.append( + ReplayNodeExpectation( + node_id=task.task_id, + objective=task.objective, + constraints=constraints, + depends_on=task.depends_on, + repository_write=bool(task.allowed_paths), + effects=( + ("repository-read", "repository-write", "process") + if task.allowed_paths + else ("repository-read", "process") + ), + allowed_paths=task.allowed_paths, + max_changed_paths=max_changed_paths, + checks=tuple(checks), + status=state.status.value, + attempt=state.attempts, + fencing_token=state.attempts, + lease_digest=outcome.lease_digest, + worktree_spec_digest=outcome.worktree_spec_digest, + base_commit=base_commit, + head_commit=state.head_commit, + failure_code=( + None if state.last_failure_class is None else state.last_failure_class.value + ), + result_digest=outcome_digest, + provider_context_digest=provider_context_digest, + ) + ) + return tuple(expectations) + def _record_run_queued( self, *, @@ -1522,7 +1693,7 @@ def _load_run(self, run_id: str) -> _LoadedRun: except ExecutionRuntimeError as error: raise RuntimeApiError(RuntimeApiFailureCode.CONFLICT) from error self._validate_worktree_evidence(events, plan) - return _LoadedRun( + loaded = _LoadedRun( request=request, intent=intent, plan=plan, @@ -1530,6 +1701,9 @@ def _load_run(self, run_id: str) -> _LoadedRun: state=state, kernel_state=kernel_state, ) + if plan.planning_mode == "generated" and kernel_state is not None: + _require_generated_kernel_binding(loaded) + return loaded def _validate_worktree_evidence( self, @@ -2385,7 +2559,16 @@ def _run_response(loaded: _LoadedRun) -> RunResponse: ), attempt=max(max(node.attempts for node in state.nodes), kernel_attempt), fencing_token=max(node.fencing_token for node in state.nodes), - retained_worktree=any(node.retained_worktree for node in state.nodes), + retained_worktree=( + any(node.retained_worktree for node in state.nodes) + or ( + kernel is not None + and ( + bool(kernel.retained_workspace_ids) + or any(node.retained_workspace_id is not None for node in kernel.tasks) + ) + ) + ), principal_id=_event_principal(state.queued_event), event_id=event.event_id, cursor=_global_position(event), @@ -2471,7 +2654,7 @@ def _run_query_item(loaded: _LoadedRun) -> RunQueryItem: failure_code=( None if node.last_failure_class is None else node.last_failure_class.value ), - retained_worktree=node.active_workspace_id is not None, + retained_worktree=node.retained_workspace_id is not None, head_commit=node.head_commit, depends_on=tasks[node.task_id].depends_on, max_attempts=tasks[node.task_id].max_attempts, @@ -2500,23 +2683,72 @@ def _run_query_item(loaded: _LoadedRun) -> RunQueryItem: def _kernel_plan(loaded: _LoadedRun) -> Plan: + plans = _kernel_plans(loaded) + if not plans: + raise RuntimeApiError(RuntimeApiFailureCode.CONFLICT) + return plans[-1] + + +def _kernel_plans(loaded: _LoadedRun) -> tuple[Plan, ...]: + plans: list[Plan] = [] + for event in loaded.events: + if event.event_type != EXECUTION_PLAN_ADMITTED: + continue + try: + plans.append(plan_from_payload(_thawed_mapping(event.payload).get("plan"))) + except (LifecycleError, TypeError, ValueError) as error: + raise RuntimeApiError(RuntimeApiFailureCode.CONFLICT) from error + kernel = loaded.kernel_state + if kernel is None: + raise RuntimeApiError(RuntimeApiFailureCode.CONFLICT) + if not plans: + if any( + value is not None + for value in (kernel.plan_id, kernel.plan_digest, kernel.plan_revision) + ): + raise RuntimeApiError(RuntimeApiFailureCode.CONFLICT) + return () + plan = plans[-1] + if ( + plan.plan_id != kernel.plan_id + or plan.plan_digest != kernel.plan_digest + or plan.plan_revision != kernel.plan_revision + ): + raise RuntimeApiError(RuntimeApiFailureCode.CONFLICT) + return tuple(plans) + + +def _kernel_goal(loaded: _LoadedRun) -> GoalSpec: event = next( - (item for item in reversed(loaded.events) if item.event_type == EXECUTION_PLAN_ADMITTED), + (item for item in loaded.events if item.event_type == EXECUTION_GOAL_ADMITTED), None, ) if event is None: raise RuntimeApiError(RuntimeApiFailureCode.CONFLICT) try: - plan = plan_from_payload(_thawed_mapping(event.payload).get("plan")) + goal = goal_from_payload(_thawed_mapping(event.payload).get("goal")) except (LifecycleError, TypeError, ValueError) as error: raise RuntimeApiError(RuntimeApiFailureCode.CONFLICT) from error if ( loaded.kernel_state is None - or plan.plan_id != loaded.kernel_state.plan_id - or plan.plan_digest != loaded.kernel_state.plan_digest + or goal.goal_id != loaded.kernel_state.goal_id + or goal.digest != loaded.kernel_state.goal_digest ): raise RuntimeApiError(RuntimeApiFailureCode.CONFLICT) - return plan + return goal + + +def _require_generated_kernel_binding(loaded: _LoadedRun) -> None: + goal = _kernel_goal(loaded) + required_checks = {check.check_id: check for check in goal.verification_checks} + if goal != _generated_goal(loaded): + raise RuntimeApiError(RuntimeApiFailureCode.CONFLICT) + for plan in _kernel_plans(loaded): + internal_checks = tuple(check for task in plan.tasks for check in task.checks) + if {check.check_id for check in internal_checks} != set(required_checks) or any( + required_checks.get(check.check_id) != check for check in internal_checks + ): + raise RuntimeApiError(RuntimeApiFailureCode.CONFLICT) def _effective_run_status(loaded: _LoadedRun) -> RunStatus: @@ -2537,6 +2769,122 @@ def _effective_run_status(loaded: _LoadedRun) -> RunStatus: return statuses[kernel.status] +def _review_terminal_index(loaded: _LoadedRun) -> int | None: + kernel = loaded.kernel_state + if kernel is not None: + if kernel.status is not ExecutionRunStatus.SUCCEEDED: + return None + return next( + ( + index + for index, event in reversed(tuple(enumerate(loaded.events))) + if event.event_id == kernel.latest_event_id + ), + None, + ) + return next( + ( + index + for index, event in reversed(tuple(enumerate(loaded.events))) + if event.event_type == RUN_SUCCEEDED + ), + None, + ) + + +def _kernel_outcome( + artifacts: ExecutionArtifactReaderPort, + digests: tuple[str, ...], +) -> tuple[str, NodeOutcomeManifest]: + try: + candidates = tuple( + digest for digest in digests if artifacts.stat(digest).media_type == OUTCOME_MEDIA_TYPE + ) + if len(candidates) != 1: + raise RuntimeApiError(RuntimeApiFailureCode.CONFLICT) + digest = candidates[0] + outcome = node_outcome_from_mapping(_artifact_mapping(artifacts, digest)) + except RuntimeApiError: + raise + except Exception as error: + raise RuntimeApiError(RuntimeApiFailureCode.CONFLICT) from error + if outcome.digest != digest: + raise RuntimeApiError(RuntimeApiFailureCode.CONFLICT) + return digest, outcome + + +def _artifact_mapping( + artifacts: ExecutionArtifactReaderPort, + digest: str, +) -> Mapping[str, object]: + try: + value = msgspec.json.decode(artifacts.get_bytes(digest, verify=True)) + except Exception as error: + raise RuntimeApiError(RuntimeApiFailureCode.CONFLICT) from error + if not isinstance(value, Mapping) or not all(isinstance(key, str) for key in value): + raise RuntimeApiError(RuntimeApiFailureCode.CONFLICT) + return cast("Mapping[str, object]", value) + + +def _artifact_integer(value: Mapping[str, object], field: str) -> int: + item = value.get(field) + if isinstance(item, bool) or not isinstance(item, int) or item < 0: + raise RuntimeApiError(RuntimeApiFailureCode.CONFLICT) + return item + + +def _artifact_number(value: Mapping[str, object], field: str) -> int | float: + item = value.get(field) + if ( + isinstance(item, bool) + or not isinstance(item, int | float) + or not isfinite(item) + or item <= 0 + ): + raise RuntimeApiError(RuntimeApiFailureCode.CONFLICT) + return item + + +def _kernel_context_constraints( + value: Mapping[str, object], + *, + accepted: tuple[str, ...], +) -> tuple[str, ...]: + raw = value.get("constraints") + if not isinstance(raw, list) or not all(isinstance(item, str) for item in raw): + raise RuntimeApiError(RuntimeApiFailureCode.CONFLICT) + constraints = tuple(cast("list[str]", raw)) + if ( + constraints[: len(accepted)] != accepted + or len(constraints) not in {len(accepted), len(accepted) + 1} + or ( + len(constraints) == len(accepted) + 1 + and ( + not constraints[-1].startswith("prior-attempt:") + or len(constraints[-1].encode("utf-8")) > 2 * 1024 + ) + ) + ): + raise RuntimeApiError(RuntimeApiFailureCode.CONFLICT) + return constraints + + +def _kernel_task_base_commit( + plan: Plan, + states: Mapping[str, ExecutionTaskState], + task_id: str, +) -> str: + task = next((item for item in plan.tasks if item.task_id == task_id), None) + if task is None: + raise RuntimeApiError(RuntimeApiFailureCode.CONFLICT) + if not task.depends_on: + return plan.base_commit + heads = {states[dependency].head_commit for dependency in task.depends_on} + if None in heads or len(heads) != 1: + raise RuntimeApiError(RuntimeApiFailureCode.CONFLICT) + return cast("str", next(iter(heads))) + + def _kernel_latest_event(loaded: _LoadedRun, state: ExecutionRunState) -> EventEnvelope: try: return next( @@ -2841,6 +3189,22 @@ def _generated_goal(loaded: _LoadedRun) -> GoalSpec: ) +def _generated_authority(plan: PlanRequest) -> ExecutionAuthority: + return ExecutionAuthority( + budget=GatewayBudget( + sum(node.budget.max_input_tokens for node in plan.nodes), + sum(node.budget.max_output_tokens for node in plan.nodes), + sum(node.budget.timeout_seconds for node in plan.nodes) * 1_000, + sum(node.budget.max_cost_microusd for node in plan.nodes), + ), + check_timeout_seconds=min(node.budget.timeout_seconds for node in plan.nodes), + max_changed_paths=min( + 10_000, + sum(node.budget.max_changed_files for node in plan.nodes), + ), + ) + + def _generated_checks(plan: PlanRequest) -> tuple[VerificationCheck, ...]: checks: dict[str, VerificationCheck] = {} for node in plan.nodes: diff --git a/src/blackcell/orchestration/execution_plan.py b/src/blackcell/orchestration/execution_plan.py index d1ac0ea..e54490d 100644 --- a/src/blackcell/orchestration/execution_plan.py +++ b/src/blackcell/orchestration/execution_plan.py @@ -114,6 +114,39 @@ class RunLifecycleStatus(StrEnum): TERMINAL_FAILURE = "terminal-failure" +@dataclass(frozen=True, slots=True) +class ExecutionAuthority: + budget: GatewayBudget + check_timeout_seconds: int + max_changed_paths: int + consumed_budget: GatewayBudget = field( + default_factory=lambda: GatewayBudget(0, 0, 0, 0), + ) + input_tokens_complete: bool = True + output_tokens_complete: bool = True + cost_microusd_complete: bool = True + + def __post_init__(self) -> None: + if ( + not isinstance(self.budget, GatewayBudget) + or not isinstance(self.consumed_budget, GatewayBudget) + or self.consumed_budget.max_input_tokens > self.budget.max_input_tokens + or self.consumed_budget.max_output_tokens > self.budget.max_output_tokens + or self.consumed_budget.max_latency_ms > self.budget.max_latency_ms + or self.consumed_budget.max_cost_microusd > self.budget.max_cost_microusd + or not isinstance(self.input_tokens_complete, bool) + or not isinstance(self.output_tokens_complete, bool) + or not isinstance(self.cost_microusd_complete, bool) + or isinstance(self.check_timeout_seconds, bool) + or not isinstance(self.check_timeout_seconds, int) + or not 1 <= self.check_timeout_seconds <= 600 + or isinstance(self.max_changed_paths, bool) + or not isinstance(self.max_changed_paths, int) + or not 0 <= self.max_changed_paths <= 10_000 + ): + raise PlanContractError() + + @dataclass(frozen=True, slots=True) class GoalSpec: goal_id: str @@ -628,6 +661,10 @@ def compile_plan( if not 1 <= len(raw_tasks) <= _MAX_TASKS: raise PlanContractError() tasks = tuple(_compile_task(goal, raw) for raw in raw_tasks) + covered_checks = {check.check_id for task in tasks for check in task.checks} + required_checks = {check.check_id for check in goal.verification_checks} + if covered_checks != required_checks: + raise PlanContractError("plan-check-coverage-incomplete") draft_digest = json_digest(cast("Mapping[str, JsonInput]", draft)) if previous is None: revision = 1 @@ -1001,6 +1038,7 @@ def _sequence(value: object) -> Sequence[object]: "PLAN_DRAFT_OUTPUT_SCHEMA", "AttemptEvidence", "AttemptRoute", + "ExecutionAuthority", "ExecutionPolicyKernel", "FailureClass", "GoalSpec", diff --git a/src/blackcell/orchestration/execution_runtime.py b/src/blackcell/orchestration/execution_runtime.py index 5af415e..22ba898 100644 --- a/src/blackcell/orchestration/execution_runtime.py +++ b/src/blackcell/orchestration/execution_runtime.py @@ -186,6 +186,8 @@ class ExecutionTaskState: last_error_signature: str | None = None same_error_count: int = 0 active_workspace_id: str | None = None + retained_workspace_id: str | None = None + blocked_reason: str | None = None evidence_event_ids: tuple[str, ...] = () last_failure_class: FailureClass | None = None last_failure_summary: str | None = None @@ -214,6 +216,7 @@ class ExecutionRunState: plan_revision: int | None status: RunLifecycleStatus tasks: tuple[ExecutionTaskState, ...] + retained_workspace_ids: tuple[str, ...] latest_event_id: str last_stream_sequence: int input_tokens: int = 0 @@ -238,7 +241,7 @@ def record(self, event: EventEnvelope) -> None: ... class ExecutionRunProjection: name = "run-kernel" - version = 5 + version = 6 def initial_state(self) -> ExecutionRunState | None: return None @@ -333,6 +336,7 @@ def apply( plan_revision=None, status=RunLifecycleStatus.ADMITTED, tasks=(), + retained_workspace_ids=(), latest_event_id=event.event_id, last_stream_sequence=event.stream_sequence, ) @@ -616,6 +620,11 @@ def apply( last_error_signature=signature, same_error_count=same_error_count, active_workspace_id=None, + retained_workspace_id=( + task.retained_workspace_id + if route is AttemptRoute.SUCCEEDED + else task.active_workspace_id + ), evidence_event_ids=(*task.evidence_event_ids, event.event_id), last_failure_class=failure_class, last_failure_summary=failure_summary, @@ -624,8 +633,16 @@ def apply( head_commit=evidence.head_commit, ), ) + retained_workspace_ids = updated.retained_workspace_ids + if ( + route is not AttemptRoute.SUCCEEDED + and task.active_workspace_id is not None + and task.active_workspace_id not in retained_workspace_ids + ): + retained_workspace_ids = (*retained_workspace_ids, task.active_workspace_id) updated = replace( updated, + retained_workspace_ids=retained_workspace_ids, input_tokens=updated.input_tokens + (input_tokens or 0), input_tokens_complete=(updated.input_tokens_complete and input_tokens is not None), output_tokens=updated.output_tokens + (output_tokens or 0), @@ -652,6 +669,7 @@ def apply( replace( task, status=TaskLifecycleStatus.BLOCKED, + blocked_reason=reason, pending_policy_decision_id=None, pending_policy_action_digest=None, pending_policy_allowed=None, @@ -692,6 +710,7 @@ def dump_state(self, state: ExecutionRunState | None) -> JsonInput: "plan_digest": state.plan_digest, "plan_version": state.plan_revision, "status": state.status.value, + "retained_workspace_ids": list(state.retained_workspace_ids), "tasks": [ { "task_id": item.task_id, @@ -703,6 +722,8 @@ def dump_state(self, state: ExecutionRunState | None) -> JsonInput: "last_error_signature": item.last_error_signature, "same_error_count": item.same_error_count, "active_workspace_id": item.active_workspace_id, + "retained_workspace_id": item.retained_workspace_id, + "blocked_reason": item.blocked_reason, "evidence_event_ids": list(item.evidence_event_ids), "last_failure_class": ( None if item.last_failure_class is None else item.last_failure_class.value @@ -750,6 +771,7 @@ def load_state(self, value: object) -> ExecutionRunState | None: "plan_digest", "plan_version", "status", + "retained_workspace_ids", "tasks", "latest_event_id", "last_stream_sequence", @@ -773,6 +795,9 @@ def load_state(self, value: object) -> ExecutionRunState | None: isinstance(plan_revision_value, bool) or not isinstance(plan_revision_value, int) ): raise ExecutionRuntimeError("invalid-execution-checkpoint") + retained_workspace_ids = _text_tuple(raw, "retained_workspace_ids") + if len(set(retained_workspace_ids)) != len(retained_workspace_ids): + raise ExecutionRuntimeError("invalid-execution-checkpoint") return ExecutionRunState( run_id=_text(raw, "run_id"), goal_id=_text(raw, "goal_id"), @@ -787,6 +812,7 @@ def load_state(self, value: object) -> ExecutionRunState | None: plan_digest=_optional_digest(raw, "plan_digest"), plan_revision=cast("int | None", plan_revision_value), status=RunLifecycleStatus(_text(raw, "status")), + retained_workspace_ids=retained_workspace_ids, tasks=tasks, latest_event_id=_text(raw, "latest_event_id"), last_stream_sequence=_integer(raw, "last_stream_sequence"), @@ -1061,6 +1087,19 @@ def exhaust_replan_budget(self, run_id: str, *, actor: str) -> ExecutionRunState reason="replan-budget-exhausted", ) + def exhaust_provider_budget(self, run_id: str, *, actor: str) -> ExecutionRunState: + """Terminate a replan before redispatch when canonical provider authority is spent.""" + + state = self.journal.rehydrate(run_id) + if state.status is not RunLifecycleStatus.REPLAN_REQUIRED: + raise ExecutionRuntimeError("execution-replan-not-required") + return self._terminate( + run_id, + RunLifecycleStatus.ESCALATED, + actor=actor, + reason="cumulative-budget-exhausted", + ) + def compile_and_admit( self, run_id: str, @@ -1068,9 +1107,12 @@ def compile_and_admit( *, actor: str, previous: Plan | None = None, + provider_budget: GatewayBudget | None = None, ) -> tuple[Plan, PlanningResult]: if request.run_id != run_id: raise ExecutionRuntimeError("execution-run-binding-mismatch") + if provider_budget is not None and not _budget_within(provider_budget, request.budget): + raise ExecutionRuntimeError("execution-provider-budget-invalid") goal = request.goal if previous is None: self.journal.append( @@ -1109,7 +1151,10 @@ def compile_and_admit( }, actor=actor, ) - result = self.provider.propose_plan(request) + provider_request = ( + request if provider_budget is None else replace(request, budget=provider_budget) + ) + result = self.provider.propose_plan(provider_request) plan = compile_plan( goal, cast("Mapping[str, object]", result.draft), @@ -1168,6 +1213,19 @@ def execute( raise ExecutionRuntimeError("execution-run-binding-mismatch") if state.status in _RUN_OUTCOME_STATUSES: return state + blocked = next( + (item for item in state.tasks if item.status is TaskLifecycleStatus.BLOCKED), + None, + ) + if blocked is not None: + if blocked.blocked_reason is None: + raise ExecutionRuntimeError("execution-blocked-task-invalid") + return self._terminate( + run_id, + RunLifecycleStatus.BLOCKED, + actor=actor, + reason=blocked.blocked_reason, + ) active = next((item for item in state.tasks if item.active_workspace_id is not None), None) if active is not None: return self._terminate( @@ -1222,7 +1280,10 @@ def execute( ) state = self.journal.rehydrate(run_id) remaining_budget = _remaining_budget(request.budget, state) - if task.allowed_paths and _provider_budget_exhausted(remaining_budget): + if task.allowed_paths and ( + _provider_usage_incomplete(state) + or _provider_budget_exhausted(remaining_budget) + ): return self._terminate( run_id, RunLifecycleStatus.ESCALATED, @@ -1237,8 +1298,10 @@ def execute( capability="repository-task", allowed_paths=task.allowed_paths, ) - decision = self.policy.authorize(goal, plan, task, action) - self._record_policy(run_id, task_id, attempt, decision, actor=actor) + decision = _pending_policy_decision(task_state, action) + if decision is None: + decision = self.policy.authorize(goal, plan, task, action) + self._record_policy(run_id, task_id, attempt, decision, actor=actor) if not decision.allowed: self.journal.append( run_id, @@ -1446,8 +1509,6 @@ def _workspace_id_from_ids(plan_id: str, task_id: str, attempt: int) -> str: def _expected_base_commit(state: ExecutionRunState, task: ExecutionTaskState) -> str: - if task.head_commit is not None: - return task.head_commit if not task.depends_on: return state.goal_base_commit dependency_heads = tuple(state.task(item).head_commit for item in task.depends_on) @@ -1475,12 +1536,50 @@ def _remaining_budget(budget: GatewayBudget, state: ExecutionRunState) -> Gatewa ) +def _budget_within(candidate: GatewayBudget, maximum: GatewayBudget) -> bool: + return ( + candidate.max_input_tokens <= maximum.max_input_tokens + and candidate.max_output_tokens <= maximum.max_output_tokens + and candidate.max_latency_ms <= maximum.max_latency_ms + and candidate.max_cost_microusd <= maximum.max_cost_microusd + ) + + +def _pending_policy_decision( + task: ExecutionTaskState, + action: ToolActionRequest, +) -> PolicyDecision | None: + decision_id = task.pending_policy_decision_id + if decision_id is None: + return None + action_digest = task.pending_policy_action_digest + allowed = task.pending_policy_allowed + reason = task.pending_policy_reason + if action_digest is None or allowed is None or reason is None: + raise ExecutionRuntimeError("execution-pending-policy-invalid") + try: + decision = PolicyDecision(allowed, reason, action_digest) + except (TypeError, ValueError) as error: + raise ExecutionRuntimeError("execution-pending-policy-invalid") from error + if decision.action_digest != action.digest or decision.decision_id != decision_id: + raise ExecutionRuntimeError("execution-pending-policy-invalid") + return decision + + def _provider_budget_exhausted(budget: GatewayBudget) -> bool: return ( budget.max_input_tokens == 0 or budget.max_output_tokens == 0 or budget.max_latency_ms == 0 ) +def _provider_usage_incomplete(state: ExecutionRunState) -> bool: + return ( + not state.input_tokens_complete + or not state.output_tokens_complete + or not state.cost_microusd_complete + ) + + def _budget_overdrawn(budget: GatewayBudget, state: ExecutionRunState) -> bool: return ( state.input_tokens > budget.max_input_tokens @@ -1532,31 +1631,30 @@ def _require_fields(value: Mapping[str, object], expected: set[str], error_code: def _task_state_from_checkpoint(value: object) -> ExecutionTaskState: raw = _mapping(value) - _require_fields( - raw, - { - "task_id", - "status", - "depends_on", - "allowed_paths", - "max_attempts", - "attempts", - "last_error_signature", - "same_error_count", - "active_workspace_id", - "evidence_event_ids", - "last_failure_class", - "last_failure_summary", - "last_artifact_digests", - "last_progress_digests", - "head_commit", - "pending_policy_decision_id", - "pending_policy_action_digest", - "pending_policy_allowed", - "pending_policy_reason", - }, - "invalid-execution-checkpoint", - ) + fields = { + "task_id", + "status", + "depends_on", + "allowed_paths", + "max_attempts", + "attempts", + "last_error_signature", + "same_error_count", + "active_workspace_id", + "retained_workspace_id", + "blocked_reason", + "evidence_event_ids", + "last_failure_class", + "last_failure_summary", + "last_artifact_digests", + "last_progress_digests", + "head_commit", + "pending_policy_decision_id", + "pending_policy_action_digest", + "pending_policy_allowed", + "pending_policy_reason", + } + _require_fields(raw, fields, "invalid-execution-checkpoint") task = ExecutionTaskState( task_id=_text(raw, "task_id"), status=TaskLifecycleStatus(_text(raw, "status")), @@ -1567,6 +1665,8 @@ def _task_state_from_checkpoint(value: object) -> ExecutionTaskState: last_error_signature=_optional_digest(raw, "last_error_signature"), same_error_count=_integer(raw, "same_error_count"), active_workspace_id=_optional_text(raw, "active_workspace_id"), + retained_workspace_id=_optional_text(raw, "retained_workspace_id"), + blocked_reason=_optional_text(raw, "blocked_reason"), evidence_event_ids=_text_tuple(raw, "evidence_event_ids"), last_failure_class=( None @@ -1592,6 +1692,8 @@ def _task_state_from_checkpoint(value: object) -> ExecutionTaskState: any(item is not None for item in pending) and task.status is not TaskLifecycleStatus.READY ): raise ExecutionRuntimeError("invalid-execution-checkpoint") + if (task.status is TaskLifecycleStatus.BLOCKED) != (task.blocked_reason is not None): + raise ExecutionRuntimeError("invalid-execution-checkpoint") if ( not 1 <= task.max_attempts <= 3 or task.attempts > task.max_attempts diff --git a/src/blackcell/orchestration/replay.py b/src/blackcell/orchestration/replay.py index 7f0dfcc..5e3272a 100644 --- a/src/blackcell/orchestration/replay.py +++ b/src/blackcell/orchestration/replay.py @@ -151,7 +151,7 @@ class ReplayCheckExpectation: check_id: str argv: tuple[str, ...] expected_exit_code: int - timeout_seconds: int + timeout_seconds: int | float @dataclass(frozen=True, slots=True) @@ -582,6 +582,7 @@ def build_review_context_from_artifacts( base_commit: str, state_digest: str, nodes: tuple[ReplayNodeExpectation, ...], + require_exact_constraints: bool = True, ) -> ReviewContext: """Construct complete review input only from a replay-verified artifact graph.""" @@ -591,7 +592,7 @@ def build_review_context_from_artifacts( if ( len(materials) != len(nodes) or any(node.status != "succeeded" for node in nodes) - or any(node.constraints != constraints for node in nodes) + or (require_exact_constraints and any(node.constraints != constraints for node in nodes)) or any(node.repository_write != ("repository-write" in node.effects) for node in nodes) or any(not node.depends_on and node.base_commit != base_commit for node in nodes) or any(not material.checks for material in materials) diff --git a/tests/integration/test_execution_runtime_production.py b/tests/integration/test_execution_runtime_production.py index a6ad71d..06be168 100644 --- a/tests/integration/test_execution_runtime_production.py +++ b/tests/integration/test_execution_runtime_production.py @@ -4,7 +4,7 @@ import subprocess import sys from collections.abc import Callable -from dataclasses import dataclass +from dataclasses import dataclass, field from datetime import UTC, datetime from pathlib import Path from typing import cast @@ -16,7 +16,11 @@ BubblewrapExecutable, BubblewrapIsolationPolicy, ) -from blackcell.adapters.execution.worktree import GitWorktreeLifecycle, WorktreeExecutionSpec +from blackcell.adapters.execution.worktree import ( + GitWorktreeLifecycle, + WorktreeExecutionSpec, + worktree_inspection_payload, +) from blackcell.adapters.models.change_provider import ( ChangeProviderError, ChangeProviderFailureCode, @@ -37,6 +41,7 @@ PlanNode, PlanRequest, ProjectRequest, + RunQueryRequest, RunRequest, ) from blackcell.kernel import ArtifactStore, CheckpointStore, EventStore, JsonInput, JsonValue @@ -44,6 +49,7 @@ from blackcell.orchestration.acceptance import ( AcceptanceCommand, AcceptanceResult, + AcceptanceStream, ) from blackcell.orchestration.changes import ( ChangeProposal, @@ -54,7 +60,9 @@ ) from blackcell.orchestration.execution_plan import ( EXECUTION_PLAN_DRAFT_SCHEMA, + ExecutionAuthority, ExecutionPolicyKernel, + FailureClass, GoalSpec, PlanningRequest, PlanningResult, @@ -98,10 +106,13 @@ def propose_plan(self, request: PlanningRequest) -> PlanningResult: @dataclass class RepairingChangeProvider: + latency_ms: int = 1 calls: int = 0 + budgets: list[GatewayBudget] = field(default_factory=list) def propose(self, call: ChangeProviderCall) -> ChangeProviderResult: self.calls += 1 + self.budgets.append(call.budget) evidence_file = next(item for item in call.context.files if item.path == "value.txt") replacement = "still-broken\n" if self.calls == 1 else "fixed\n" proposal = ChangeProposal( @@ -125,7 +136,7 @@ def propose(self, call: ChangeProviderCall) -> ChangeProviderResult: model_id="recorded-integration", input_tokens=1, output_tokens=1, - latency_ms=1, + latency_ms=self.latency_ms, cost_microusd=0, completed_at=NOW, ) @@ -134,13 +145,80 @@ def propose(self, call: ChangeProviderCall) -> ChangeProviderResult: @dataclass class FailingChangeProvider: calls: int = 0 + budgets: list[GatewayBudget] = field(default_factory=list) def propose(self, call: ChangeProviderCall) -> ChangeProviderResult: - del call self.calls += 1 + self.budgets.append(call.budget) raise ChangeProviderError(ChangeProviderFailureCode.INVALID_GATEWAY_RESULT) +@dataclass +class ExpandingChangeProvider: + paths: tuple[str, ...] + calls: int = 0 + + def propose(self, call: ChangeProviderCall) -> ChangeProviderResult: + path = self.paths[self.calls] + self.calls += 1 + evidence_file = next(item for item in call.context.files if item.path == path) + proposal = ChangeProposal( + proposal_id=f"expand-{self.calls}", + evidence_digest=call.context.digest, + operations=( + FileChange( + TextOperation.REPLACE, + path, + evidence_file.content_digest, + f"changed-{self.calls}\n", + ), + ), + summary="Attempt to expand the cumulative changed-file set.", + ) + return ChangeProviderResult( + proposal=proposal, + provider_output_digest=proposal.digest, + profile_id="integration-code", + adapter_id="recorded-integration", + model_id="recorded-integration", + input_tokens=1, + output_tokens=1, + latency_ms=1, + cost_microusd=0, + completed_at=NOW, + ) + + +@dataclass +class FailingAcceptance: + lifecycle: GitWorktreeLifecycle + calls: int = 0 + + def run( + self, + command: AcceptanceCommand, + spec: WorktreeExecutionSpec, + *, + cancel_requested: Callable[[], bool] | None = None, + ) -> AcceptanceResult: + assert cancel_requested is None or not cancel_requested() + self.calls += 1 + inspection_digest = json_digest(worktree_inspection_payload(self.lifecycle.inspect(spec))) + return AcceptanceResult( + check_id=command.check_id, + command_digest=command.digest, + worktree_spec_digest=spec.digest, + isolation_policy_digest=json_digest({"kind": "recorded-integration"}), + inspection_before_digest=inspection_digest, + inspection_after_digest=inspection_digest, + return_code=1, + expected_exit_code=command.expected_exit_code, + passed=False, + stdout=AcceptanceStream(b""), + stderr=AcceptanceStream(b"intentional failure\n"), + ) + + class UnusedAcceptance: def run( self, @@ -153,6 +231,84 @@ def run( raise AssertionError("provider failure must precede acceptance execution") +@dataclass +class RecordingAcceptance: + delegate: BubblewrapAcceptanceRunner + commands: list[AcceptanceCommand] = field(default_factory=list) + specs: list[WorktreeExecutionSpec] = field(default_factory=list) + + def run( + self, + command: AcceptanceCommand, + spec: WorktreeExecutionSpec, + *, + cancel_requested: Callable[[], bool] | None = None, + ) -> AcceptanceResult: + self.commands.append(command) + self.specs.append(spec) + return self.delegate.run(command, spec, cancel_requested=cancel_requested) + + +@dataclass +class ManualClock: + elapsed_ms: int = 0 + invalid: bool = False + samples: tuple[float, ...] = () + sample_index: int = 0 + + def __call__(self) -> float: + if self.invalid: + return float("nan") + if self.sample_index < len(self.samples): + value = self.samples[self.sample_index] + self.sample_index += 1 + return value + return self.elapsed_ms / 1_000 + + def advance(self, elapsed_ms: int) -> None: + self.elapsed_ms += elapsed_ms + + +@dataclass +class TimedAcceptance: + lifecycle: GitWorktreeLifecycle + clock: ManualClock + durations_ms: tuple[int, ...] + error: Exception | None = None + invalidate_clock: bool = False + commands: list[AcceptanceCommand] = field(default_factory=list) + + def run( + self, + command: AcceptanceCommand, + spec: WorktreeExecutionSpec, + *, + cancel_requested: Callable[[], bool] | None = None, + ) -> AcceptanceResult: + assert cancel_requested is None or not cancel_requested() + duration_ms = self.durations_ms[len(self.commands)] + self.commands.append(command) + self.clock.advance(duration_ms) + if self.invalidate_clock: + self.clock.invalid = True + if self.error is not None: + raise self.error + inspection_digest = json_digest(worktree_inspection_payload(self.lifecycle.inspect(spec))) + return AcceptanceResult( + check_id=command.check_id, + command_digest=command.digest, + worktree_spec_digest=spec.digest, + isolation_policy_digest=json_digest({"kind": "timed-integration"}), + inspection_before_digest=inspection_digest, + inspection_after_digest=inspection_digest, + return_code=command.expected_exit_code, + expected_exit_code=command.expected_exit_code, + passed=True, + stdout=AcceptanceStream(b""), + stderr=AcceptanceStream(b""), + ) + + def test_executor_preserves_unknown_provider_usage_and_rejects_replayed_decision( tmp_path: Path, ) -> None: @@ -219,6 +375,24 @@ def test_executor_preserves_unknown_provider_usage_and_rejects_replayed_decision decision = ExecutionPolicyKernel().authorize(goal, plan, task, action) provider = FailingChangeProvider() budget = GatewayBudget(32_000, 4_096, 30_000, 0) + authority_calls = 0 + + def authority_for_run(run_id: str) -> ExecutionAuthority: + nonlocal authority_calls + assert run_id == "run-provider-failure" + authority_calls += 1 + if authority_calls == 1: + return ExecutionAuthority(budget, 10, 1) + return ExecutionAuthority( + budget=budget, + check_timeout_seconds=10, + max_changed_paths=1, + consumed_budget=GatewayBudget(32_000, 4_096, 0, 0), + input_tokens_complete=False, + output_tokens_complete=False, + cost_microusd_complete=False, + ) + executor = ProductionAttemptExecutor( repository_root=repository, isolation_root=isolation, @@ -235,6 +409,7 @@ def test_executor_preserves_unknown_provider_usage_and_rejects_replayed_decision stderr_limit_bytes=64 * 1024, ), worktrees=worktrees, + authority_for_run=authority_for_run, ) evidence = executor.execute( @@ -252,6 +427,7 @@ def test_executor_preserves_unknown_provider_usage_and_rejects_replayed_decision ) assert provider.calls == 1 + assert provider.budgets == [budget] assert evidence.input_tokens is None assert evidence.output_tokens is None assert evidence.cost_microusd is None @@ -271,6 +447,376 @@ def test_executor_preserves_unknown_provider_usage_and_rejects_replayed_decision ) assert provider.calls == 1 + second_action = ToolActionRequest( + run_id="run-provider-failure", + plan_id=plan.plan_id, + task_id=task.task_id, + attempt=2, + capability="repository-task", + allowed_paths=task.allowed_paths, + ) + blocked = executor.execute( + run_id="run-provider-failure", + goal=goal, + plan=plan, + task=task, + attempt=2, + workspace_id="workspace-provider-failure-2", + base_commit=base_commit, + prior_failure_class=evidence.failure_class, + prior_failure_summary=evidence.failure_summary, + policy_decision=ExecutionPolicyKernel().authorize( + goal, + plan, + task, + second_action, + ), + remaining_budget=budget, + ) + + assert blocked.failure_class is FailureClass.POLICY + assert blocked.failure_summary == "execution-cumulative-budget-exhausted" + assert provider.calls == 1 + + +def test_executor_blocks_cumulative_changed_path_expansion_across_retries( + tmp_path: Path, +) -> None: + git = _executables("git")["git"] + repository = tmp_path / "repository" + repository.mkdir() + (repository / "first.txt").write_text("first\n", encoding="utf-8") + (repository / "second.txt").write_text("second\n", encoding="utf-8") + _git(repository, git, "init", "--initial-branch=main") + _git(repository, git, "add", "first.txt", "second.txt") + _git( + repository, + git, + "-c", + "user.name=Fixture", + "-c", + "user.email=fixture@example.invalid", + "commit", + "-m", + "fixture", + ) + base_commit = _git_text(repository, git, "rev-parse", "HEAD") + isolation = tmp_path / "worktrees" + isolation.mkdir(mode=0o700) + artifacts = ArtifactStore(tmp_path / "artifacts", database_path=tmp_path / "kernel.sqlite3") + worktrees = GitWorktreeLifecycle(git_executable=git) + goal = GoalSpec( + goal_id="goal-cumulative-paths", + project_id="project-cumulative-paths", + intent_id="intent-cumulative-paths", + objective="Keep cumulative changes within one file.", + base_commit=base_commit, + constraints=(), + allowed_paths=("first.txt", "second.txt"), + verification_checks=(VerificationCheck("check", ("true",)),), + max_attempts=2, + ) + plan = compile_plan( + goal, + { + "schema_version": EXECUTION_PLAN_DRAFT_SCHEMA, + "tasks": [ + { + "task_id": "repair", + "objective": "Exercise cumulative authority.", + "depends_on": [], + "allowed_paths": ["first.txt", "second.txt"], + "checks": ["check"], + } + ], + }, + ) + task = plan.tasks[0] + provider = ExpandingChangeProvider(("first.txt", "second.txt")) + acceptance = FailingAcceptance(worktrees) + budget = GatewayBudget(32_000, 4_096, 30_000, 0) + authority = ExecutionAuthority(budget, 10, 1) + executor = ProductionAttemptExecutor( + repository_root=repository, + isolation_root=isolation, + artifacts=artifacts, + change_provider=provider, + acceptance=acceptance, + policy=ExecutionPolicy( + worker_id="integration-worker", + classification=DataClassification.PRIVATE, + locality=LocalityPolicy.REMOTE_ALLOWED, + provider_budget=budget, + check_timeout_seconds=30, + stdout_limit_bytes=64 * 1024, + stderr_limit_bytes=64 * 1024, + max_changed_paths=8, + remove_successful_worktrees=False, + ), + worktrees=worktrees, + authority_for_run=lambda _: authority, + ) + + first_action = ToolActionRequest( + run_id="run-cumulative-paths", + plan_id=plan.plan_id, + task_id=task.task_id, + attempt=1, + capability="repository-task", + allowed_paths=task.allowed_paths, + ) + first = executor.execute( + run_id="run-cumulative-paths", + goal=goal, + plan=plan, + task=task, + attempt=1, + workspace_id="workspace-cumulative-paths-1", + base_commit=base_commit, + prior_failure_class=None, + prior_failure_summary=None, + policy_decision=ExecutionPolicyKernel().authorize( + goal, + plan, + task, + first_action, + ), + remaining_budget=budget, + ) + second_action = ToolActionRequest( + run_id="run-cumulative-paths", + plan_id=plan.plan_id, + task_id=task.task_id, + attempt=2, + capability="repository-task", + allowed_paths=task.allowed_paths, + ) + second = executor.execute( + run_id="run-cumulative-paths", + goal=goal, + plan=plan, + task=task, + attempt=2, + workspace_id="workspace-cumulative-paths-2", + base_commit=first.head_commit, + prior_failure_class=first.failure_class, + prior_failure_summary=first.failure_summary, + policy_decision=ExecutionPolicyKernel().authorize( + goal, + plan, + task, + second_action, + ), + remaining_budget=budget, + ) + + assert first.failure_class is FailureClass.LOGIC_BUG + assert second.failure_class is FailureClass.POLICY + assert second.failure_summary == "execution-cumulative-path-limit-exceeded" + assert second.head_commit == first.head_commit + assert provider.calls == 2 + assert acceptance.calls == 1 + assert worktrees.changed_paths_between( + repository, + base_commit=base_commit, + head_commit=second.head_commit, + ) == ("first.txt",) + + +@pytest.mark.parametrize( + ( + "remaining_latency_ms", + "durations_ms", + "expected_timeouts", + "runner_error", + "invalid_start_clock", + "invalid_finish_clock", + "extreme_clock_span", + "expected_failure", + ), + ( + (5_000, (1_000, 500), (3.0, 2.0), None, False, False, False, None), + ( + 2_500, + (500,), + (0.5,), + None, + False, + False, + False, + "execution-cumulative-budget-exhausted", + ), + ( + 2_500, + (600,), + (0.5,), + None, + False, + False, + False, + "execution-cumulative-budget-exhausted", + ), + ( + 5_000, + (1_000, 2_100), + (3.0, 2.0), + None, + False, + False, + False, + "execution-cumulative-budget-exhausted", + ), + ( + 5_000, + (700,), + (3.0,), + OSError("acceptance runner failed"), + False, + False, + False, + "execution-acceptance-runner-failed", + ), + (5_000, (), (), None, True, False, False, "execution-clock-invalid"), + (5_000, (100,), (3.0,), None, False, True, False, "execution-clock-invalid"), + (5_000, (100,), (3.0,), None, False, False, True, "execution-clock-invalid"), + ), +) +def test_executor_debits_acceptance_time_from_durable_latency_authority( + tmp_path: Path, + *, + remaining_latency_ms: int, + durations_ms: tuple[int, ...], + expected_timeouts: tuple[float, ...], + runner_error: Exception | None, + invalid_start_clock: bool, + invalid_finish_clock: bool, + extreme_clock_span: bool, + expected_failure: str | None, +) -> None: + git = _executables("git")["git"] + repository = tmp_path / "repository" + repository.mkdir() + (repository / "value.txt").write_text("broken\n", encoding="utf-8") + _git(repository, git, "init", "--initial-branch=main") + _git(repository, git, "add", "value.txt") + _git( + repository, + git, + "-c", + "user.name=Fixture", + "-c", + "user.email=fixture@example.invalid", + "commit", + "-m", + "fixture", + ) + base_commit = _git_text(repository, git, "rev-parse", "HEAD") + isolation = tmp_path / "worktrees" + isolation.mkdir(mode=0o700) + artifacts = ArtifactStore(tmp_path / "artifacts", database_path=tmp_path / "kernel.sqlite3") + worktrees = GitWorktreeLifecycle(git_executable=git) + checks = ( + VerificationCheck("check-first", ("true",)), + VerificationCheck("check-second", ("true",)), + ) + goal = GoalSpec( + goal_id="goal-latency-authority", + project_id="project-latency-authority", + intent_id="intent-latency-authority", + objective="Debit provider and acceptance latency cumulatively.", + base_commit=base_commit, + constraints=(), + allowed_paths=("value.txt",), + verification_checks=checks, + ) + plan = compile_plan( + goal, + { + "schema_version": EXECUTION_PLAN_DRAFT_SCHEMA, + "tasks": [ + { + "task_id": "repair", + "objective": "Exercise cumulative latency authority.", + "depends_on": [], + "allowed_paths": ["value.txt"], + "checks": [check.check_id for check in checks], + } + ], + }, + ) + task = plan.tasks[0] + action = ToolActionRequest( + run_id="run-latency-authority", + plan_id=plan.plan_id, + task_id=task.task_id, + attempt=1, + capability="repository-task", + allowed_paths=task.allowed_paths, + ) + total_budget = GatewayBudget(32_000, 4_096, 10_000, 0) + clock = ManualClock( + invalid=invalid_start_clock, + samples=(-1e308, 1e308) if extreme_clock_span else (), + ) + acceptance = TimedAcceptance( + worktrees, + clock, + durations_ms, + runner_error, + invalid_finish_clock, + ) + executor = ProductionAttemptExecutor( + repository_root=repository, + isolation_root=isolation, + artifacts=artifacts, + change_provider=RepairingChangeProvider(latency_ms=2_000), + acceptance=acceptance, + policy=ExecutionPolicy( + worker_id="integration-worker", + classification=DataClassification.PRIVATE, + locality=LocalityPolicy.REMOTE_ALLOWED, + provider_budget=total_budget, + check_timeout_seconds=30, + stdout_limit_bytes=64 * 1024, + stderr_limit_bytes=64 * 1024, + max_changed_paths=1, + remove_successful_worktrees=False, + ), + worktrees=worktrees, + authority_for_run=lambda _: ExecutionAuthority(total_budget, 10, 1), + clock=clock, + ) + + evidence = executor.execute( + run_id="run-latency-authority", + goal=goal, + plan=plan, + task=task, + attempt=1, + workspace_id="workspace-latency-authority-1", + base_commit=base_commit, + prior_failure_class=None, + prior_failure_summary=None, + policy_decision=ExecutionPolicyKernel().authorize(goal, plan, task, action), + remaining_budget=GatewayBudget(32_000, 4_096, remaining_latency_ms, 0), + ) + + assert tuple(command.timeout_seconds for command in acceptance.commands) == expected_timeouts + expected_latency_ms = ( + 2_000 + if invalid_start_clock + else remaining_latency_ms + if invalid_finish_clock or extreme_clock_span + else 2_000 + sum(durations_ms) + ) + assert evidence.latency_ms == expected_latency_ms + assert evidence.required_checks_passed is (expected_failure is None) + if expected_failure is None: + assert evidence.failure_class is None + else: + assert evidence.failure_class is FailureClass.POLICY + assert evidence.failure_summary == expected_failure + @pytest.mark.skipif(sys.platform != "linux", reason="Bubblewrap execution runtime is Linux-only") def test_production_kernel_repairs_real_worktree_then_verifies_in_sandbox( @@ -327,8 +873,11 @@ def test_production_kernel_repairs_real_worktree_then_verifies_in_sandbox( intent_id="intent-integration", project_id="project-integration", objective="Repair and verify the fixture repository.", - constraints=("Only value.txt may change.",), - assumptions=(), + constraints=( + "Run the admitted check.", + "Only value.txt may change.", + ), + assumptions=("The repository base is immutable.",), unresolved_questions=(), idempotency_key="intent-integration", ), @@ -382,14 +931,19 @@ def test_production_kernel_repairs_real_worktree_then_verifies_in_sandbox( ) generated = runtime.next_generated_run() assert generated is not None - acceptance = BubblewrapAcceptanceRunner( - BubblewrapIsolationPolicy( - (BubblewrapExecutable("python", system_python),), - ), - worktrees, - bubblewrap_executable=executables["bwrap"], - prlimit_executable=executables["prlimit"], - probe_executable=executables["true"], + assert generated.authority.check_timeout_seconds == 10 + assert generated.authority.max_changed_paths == 1 + assert generated.authority.budget == GatewayBudget(32_000, 4_096, 10_000, 0) + acceptance = RecordingAcceptance( + BubblewrapAcceptanceRunner( + BubblewrapIsolationPolicy( + (BubblewrapExecutable("python", system_python),), + ), + worktrees, + bubblewrap_executable=executables["bwrap"], + prlimit_executable=executables["prlimit"], + probe_executable=executables["true"], + ) ) provider = RepairingChangeProvider() executor = ProductionAttemptExecutor( @@ -403,11 +957,13 @@ def test_production_kernel_repairs_real_worktree_then_verifies_in_sandbox( classification=DataClassification.PRIVATE, locality=LocalityPolicy.REMOTE_ALLOWED, provider_budget=GatewayBudget(32_000, 4_096, 30_000, 0), - check_timeout_seconds=10, + check_timeout_seconds=30, stdout_limit_bytes=64 * 1024, stderr_limit_bytes=64 * 1024, + max_changed_paths=8, ), worktrees=worktrees, + authority_for_run=runtime.generated_execution_authority, ) recorder = TraceRecorder() journal = EventBackedExecutionRunJournal( @@ -430,7 +986,9 @@ def test_production_kernel_repairs_real_worktree_then_verifies_in_sandbox( } ) kernel = ProductionExecution( - ExecutionCoordinator(journal, planner, executor, ExecutionPolicyKernel()) + ExecutionCoordinator(journal, planner, executor, ExecutionPolicyKernel()), + authority_for_run=runtime.generated_execution_authority, + goal_for_run=runtime.generated_execution_goal, ) request = PlanningRequest( goal=generated.goal, @@ -450,10 +1008,20 @@ def test_production_kernel_repairs_real_worktree_then_verifies_in_sandbox( assert state == rehydrated assert state.status is RunLifecycleStatus.SUCCEEDED + assert acceptance.commands + assert all(0 < command.timeout_seconds < 10 for command in acceptance.commands) + assert acceptance.specs + assert all(spec.max_changed_paths == 1 for spec in acceptance.specs) task = state.task("repair") assert task.status is TaskLifecycleStatus.SUCCEEDED assert task.attempts == 2 + assert task.retained_workspace_id is not None assert provider.calls == 2 + assert provider.budgets[0] == GatewayBudget(31_999, 4_095, 9_999, 0) + assert provider.budgets[1].max_input_tokens == 31_998 + assert provider.budgets[1].max_output_tokens == 4_094 + assert 0 < provider.budgets[1].max_latency_ms <= 9_998 + assert provider.budgets[1].max_cost_microusd == 0 assert task.head_commit is not None assert ( _git_text(repository, executables["git"], "show", f"{task.head_commit}:value.txt") @@ -481,9 +1049,26 @@ def test_production_kernel_repairs_real_worktree_then_verifies_in_sandbox( assert any(item.attributes.get("attempt") == 2 for item in traces) replay = runtime.replay_run("run-integration") assert replay.run.status == "succeeded" + assert replay.run.retained_worktree assert replay.artifact_integrity == "verified" assert replay.artifacts assert not replay.findings + query = runtime.query_runs( + RunQueryRequest( + schema_version="run-query-request/v1", + run_ids=("run-integration",), + ) + ) + assert query.runs[0].nodes[0].retained_worktree + assert runtime.review_run_ids() == ("run-integration",) + candidate = runtime.review_candidate("run-integration") + assert runtime.review_candidates() == (candidate,) + context = runtime.prepare_review_context(candidate) + assert context.state_digest == candidate.state_digest + assert context.artifact_evidence_digest == candidate.artifact_evidence_digest + assert context.acceptance.constraints == generated.goal.constraints + assert tuple(node.node_id for node in context.acceptance.nodes) == ("repair",) + assert all(check.passed for node in context.acceptance.nodes for check in node.checks) corrupted = replay.artifacts[0] artifacts.path_for(corrupted.digest).write_bytes(b"corrupted-after-verification") diff --git a/tests/unit/test_execution_process.py b/tests/unit/test_execution_process.py index b6b6c89..2e529ff 100644 --- a/tests/unit/test_execution_process.py +++ b/tests/unit/test_execution_process.py @@ -6,7 +6,7 @@ import stat import subprocess from collections.abc import Callable, Iterable -from dataclasses import replace +from dataclasses import dataclass, replace from pathlib import Path from threading import Event from types import FrameType @@ -26,6 +26,7 @@ from blackcell.bootstrap.execution_worker import ExecutionWorker, ExecutionWorkerCycleResult from blackcell.bootstrap.process import main from blackcell.bootstrap.runtime_service import ( + GeneratedRun, RuntimeService, WorktreeMaintenanceReport, ) @@ -39,6 +40,7 @@ ExecutionProviderAdapter, RuntimeProcessConfig, ) +from blackcell.gateway import GatewayBudget from blackcell.interfaces.http import ( AcceptanceCheck, IntentRequest, @@ -53,6 +55,13 @@ MAX_CHANGE_CONTEXT_BYTES, MAX_CHANGE_PROPOSAL_BYTES, ) +from blackcell.orchestration.execution_plan import ( + ExecutionAuthority, + GoalSpec, + PlanningRequest, + RunLifecycleStatus, + VerificationCheck, +) TOKEN = "Runtime-worker_process-token.0123456789-ABCDEFG" CONFIGURATION_DIGEST = "sha256:" + "a" * 64 @@ -79,9 +88,17 @@ class RecordingRuntime: def __init__(self, order: list[str]) -> None: self.order = order - def next_generated_run(self) -> None: + def next_generated_run(self) -> GeneratedRun | None: return None + def generated_execution_authority(self, run_id: str) -> ExecutionAuthority: + del run_id + raise AssertionError("no generated execution authority is available") + + def generated_execution_goal(self, run_id: str) -> GoalSpec: + del run_id + raise AssertionError("no generated execution goal is available") + def should_cancel_generated_run(self, run_id: str) -> bool: return False @@ -120,6 +137,26 @@ def has_mutation_capacity(self) -> bool: return self.available +@dataclass +class CapturingGeneratedExecution: + request: PlanningRequest | None = None + + def process(self, request: PlanningRequest, *, actor: str) -> object: + assert actor == "execution-worker.test" + self.request = request + return type("GeneratedState", (), {"status": RunLifecycleStatus.SUCCEEDED})() + + +class OneGeneratedRuntime(RecordingRuntime): + def __init__(self, generated: GeneratedRun) -> None: + super().__init__([]) + self.generated = generated + + def next_generated_run(self) -> GeneratedRun | None: + generated, self.generated = self.generated, None + return generated + + def test_execution_process_reconciles_then_runs_once_against_shared_storage(tmp_path: Path) -> None: config = _config(tmp_path) @@ -131,6 +168,48 @@ def test_execution_process_reconciles_then_runs_once_against_shared_storage(tmp_ assert config.security.paths.artifact_root.is_dir() +def test_generated_execution_intersects_admitted_and_worker_budgets(tmp_path: Path) -> None: + config = _config(tmp_path) + assert config.execution_worker is not None + goal = GoalSpec( + goal_id="goal-bounded", + project_id="project-bounded", + intent_id="intent-bounded", + objective="Exercise admitted generated-run bounds.", + base_commit="a" * 40, + constraints=(), + allowed_paths=("value.txt",), + verification_checks=(VerificationCheck("check", ("true",)),), + ) + generated = GeneratedRun( + "run-bounded", + goal, + ExecutionAuthority( + budget=GatewayBudget(1_000, 500, 3_000, 0), + check_timeout_seconds=3, + max_changed_paths=1, + ), + ) + execution = CapturingGeneratedExecution() + process = ExecutionWorkerProcess( + RecordingCoordinator(("idle",)), + OneGeneratedRuntime(generated), + config, + execution=cast("Any", execution), + ) + + result = process._run_generated_once( + config.execution_worker, + actor="execution-worker.test", + ) + + assert result is not None + assert result.status == "node-succeeded" + assert execution.request is not None + assert execution.request.goal == goal + assert execution.request.budget == generated.authority.budget + + def test_execution_process_composes_codex_caps_from_change_wire_contracts( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, diff --git a/tests/unit/test_execution_runtime.py b/tests/unit/test_execution_runtime.py index aa57a4c..0ec4222 100644 --- a/tests/unit/test_execution_runtime.py +++ b/tests/unit/test_execution_runtime.py @@ -27,8 +27,19 @@ ModelResponse, RoutingDecision, ) -from blackcell.interfaces.http import CancelRunRequest, RunQueryRequest -from blackcell.kernel import CheckpointStore, EventEnvelope, EventStore, JsonValue +from blackcell.interfaces.http import ( + CancelRunRequest, + RunQueryRequest, + RuntimeApiError, + RuntimeApiFailureCode, +) +from blackcell.kernel import ( + CheckpointStore, + EventEnvelope, + EventStore, + JsonValue, + ProjectionCheckpoint, +) from blackcell.kernel._json import json_digest, thaw_json from blackcell.orchestration.execution_plan import ( EXECUTION_ATTEMPT_EVIDENCE_SCHEMA, @@ -44,6 +55,7 @@ EXECUTION_TASK_VERIFYING, AttemptEvidence, AttemptRoute, + ExecutionAuthority, ExecutionPolicyKernel, FailureClass, GoalSpec, @@ -59,6 +71,7 @@ ToolActionRequest, VerificationCheck, compile_plan, + goal_from_payload, goal_payload, plan_payload, ) @@ -144,6 +157,26 @@ def test_plan_compiler_is_deterministic_compositional_and_schema_bound() -> None assert revised.plan_id != first.plan_id +def test_goal_payload_remains_replay_compatible() -> None: + goal = _goal() + payload = goal_payload(goal) + + assert set(payload) == { + "schema_version", + "goal_id", + "project_id", + "intent_id", + "objective", + "base_commit", + "constraints", + "allowed_paths", + "verification_checks", + "max_attempts", + "same_error_limit", + } + assert goal_from_payload(payload) == goal + + def test_plan_compiler_rejects_provider_scope_escalation_cycles_and_parallel_writers() -> None: goal = _goal(allowed_paths=("src", "tests")) with pytest.raises(PlanContractError, match="plan-path-outside-goal"): @@ -152,6 +185,18 @@ def test_plan_compiler_rejects_provider_scope_escalation_cycles_and_parallel_wri untrusted_check["checks"] = ["provider-command"] with pytest.raises(PlanContractError, match="plan-check-outside-goal"): compile_plan(goal, _draft(tasks=(untrusted_check,))) + multi_check_goal = replace( + goal, + verification_checks=( + *goal.verification_checks, + VerificationCheck("check-second", ("pytest", "tests", "-q")), + ), + ) + with pytest.raises(PlanContractError, match="plan-check-coverage-incomplete"): + compile_plan( + multi_check_goal, + _draft(tasks=(_task("incomplete-checks", allowed_paths=()),)), + ) cyclic = _draft( tasks=( @@ -212,6 +257,9 @@ def test_runtime_executes_one_bounded_repair_then_succeeds_from_snapshot_tail( artifact_digests=(DIGEST,), progress_digests=(DIGEST,), head_commit=FAILED_HEAD, + input_tokens=0, + output_tokens=0, + cost_microusd=0, ), _success(), ) @@ -239,11 +287,12 @@ def test_runtime_executes_one_bounded_repair_then_succeeds_from_snapshot_tail( assert state.task("implement").status is TaskLifecycleStatus.SUCCEEDED assert state.task("implement").attempts == 2 assert state.task("implement").head_commit == SUCCESS_HEAD + assert state.task("implement").retained_workspace_id is not None assert state.input_tokens == 0 - assert state.input_tokens_complete is False - assert state.cost_microusd_complete is False + assert state.input_tokens_complete is True + assert state.cost_microusd_complete is True assert len(set(executor.workspaces)) == 2 - assert executor.base_commits == [BASE_COMMIT, FAILED_HEAD] + assert executor.base_commits == [BASE_COMMIT, BASE_COMMIT] assert executor.decisions and all(item.allowed for item in executor.decisions) events = journal.events("run-repair-success") policy_positions = [ @@ -349,16 +398,28 @@ def test_existing_public_daemon_admits_and_processes_generated_plan(tmp_path: Pa run_id=run.run_id, ) task = _task("verify", allowed_paths=()) - task["checks"] = ["verify-pass"] + task["checks"] = ["inspect-pass", "verify-pass"] + provider = _Provider(_draft(tasks=(task,))) kernel = ProductionExecution( ExecutionCoordinator( EventBackedExecutionRunJournal(events, CheckpointStore(database)), - _Provider(_draft(tasks=(task,))), + provider, _Executor((_success(),)), ExecutionPolicyKernel(), - ) + ), + authority_for_run=runtime.generated_execution_authority, + goal_for_run=runtime.generated_execution_goal, ) + with pytest.raises(ExecutionRuntimeError, match="execution-run-binding-mismatch"): + kernel.process( + replace( + request, + goal=replace(request.goal, objective="Substitute a different execution goal."), + ), + actor="daemon:worker", + ) + assert provider.calls == 0 state = kernel.process(request, actor="daemon:worker") query = runtime.query_runs( RunQueryRequest( @@ -371,14 +432,313 @@ def test_existing_public_daemon_admits_and_processes_generated_plan(tmp_path: Pa assert runtime.next_generated_run() is None assert runtime.inspect_run(run.run_id).status == "succeeded" assert query.runs[0].usage is not None - assert query.runs[0].usage.input_tokens_complete is False - assert query.runs[0].usage.max_input_tokens == request.budget.max_input_tokens + assert query.runs[0].usage.input_tokens_complete is True + assert query.runs[0].usage.max_input_tokens == generated.authority.budget.max_input_tokens assert query.runs[0].nodes[0].max_attempts == request.goal.max_attempts assert {event.stream_id for event in events.read_stream(f"run:{run.run_id}")} == { f"run:{run.run_id}" } +def test_generated_runtime_rejects_successful_substituted_kernel_goal(tmp_path: Path) -> None: + repository = _repository(tmp_path) + database = tmp_path / "public-generated-substitution.sqlite3" + events = EventStore(database) + runtime = RuntimeService(events, repository) + intent = msgspec.structs.replace(_intent(), unresolved_questions=()) + bounds = msgspec.structs.replace(_plan(repository), planning_mode="generated") + run = _run() + runtime.register_project(_project(repository), principal_id="client:test") + runtime.accept_intent(intent, principal_id="client:test") + runtime.accept_plan(bounds, principal_id="client:test") + runtime.submit_run(run, principal_id="client:test") + generated = runtime.next_generated_run() + assert generated is not None + task = _task("verify", allowed_paths=()) + task["checks"] = ["inspect-pass", "verify-pass"] + request = PlanningRequest( + goal=replace(generated.goal, objective="Substitute a different execution goal."), + classification=DataClassification.PRIVATE, + locality=LocalityPolicy.REMOTE_ALLOWED, + budget=generated.authority.budget, + estimated_input_tokens=1_000, + correlation_id=run.run_id, + run_id=run.run_id, + ) + kernel = ProductionExecution( + ExecutionCoordinator( + EventBackedExecutionRunJournal(events, CheckpointStore(database)), + _Provider(_draft(tasks=(task,))), + _Executor((_success(),)), + ExecutionPolicyKernel(), + ) + ) + + assert kernel.process(request, actor="daemon:worker").status is RunLifecycleStatus.SUCCEEDED + with pytest.raises(RuntimeApiError) as caught: + runtime.inspect_run(run.run_id) + assert caught.value.code is RuntimeApiFailureCode.CONFLICT + + +def test_generated_runtime_rejects_substituted_kernel_check_definition(tmp_path: Path) -> None: + repository = _repository(tmp_path) + database = tmp_path / "public-generated-check-substitution.sqlite3" + events = EventStore(database) + runtime = RuntimeService(events, repository) + intent = msgspec.structs.replace(_intent(), unresolved_questions=()) + bounds = msgspec.structs.replace(_plan(repository), planning_mode="generated") + run = _run() + runtime.register_project(_project(repository), principal_id="client:test") + runtime.accept_intent(intent, principal_id="client:test") + runtime.accept_plan(bounds, principal_id="client:test") + runtime.submit_run(run, principal_id="client:test") + generated = runtime.next_generated_run() + assert generated is not None + + task = _task("verify", allowed_paths=()) + task["checks"] = ["inspect-pass", "verify-pass"] + draft = _draft(tasks=(task,)) + admitted = compile_plan(generated.goal, draft) + canonical_check = admitted.tasks[0].checks[0] + substituted_check = replace( + canonical_check, + argv=("python", "-c", "raise SystemExit(0)"), + ) + substituted_plan = replace( + admitted, + tasks=( + replace( + admitted.tasks[0], + checks=(substituted_check, *admitted.tasks[0].checks[1:]), + ), + ), + ) + journal = EventBackedExecutionRunJournal(events, CheckpointStore(database)) + budget = generated.authority.budget + journal.append( + run.run_id, + EXECUTION_GOAL_ADMITTED, + { + "goal_id": generated.goal.goal_id, + "goal_digest": generated.goal.digest, + "goal": goal_payload(generated.goal), + "classification": DataClassification.PRIVATE.value, + "locality": LocalityPolicy.REMOTE_ALLOWED.value, + "budget": { + "max_input_tokens": budget.max_input_tokens, + "max_output_tokens": budget.max_output_tokens, + "max_latency_ms": budget.max_latency_ms, + "max_cost_microusd": budget.max_cost_microusd, + }, + }, + actor="daemon:planner", + ) + draft_digest = json_digest(cast("dict[str, JsonValue]", draft)) + journal.append( + run.run_id, + EXECUTION_PLAN_DRAFT_RECEIVED, + { + "draft_digest": draft_digest, + "provider_output_digest": draft_digest, + "profile_id": "recorded-plan", + "adapter_id": "recorded-plan", + "model_id": "recorded-plan", + "input_tokens": 1, + "output_tokens": 1, + "latency_ms": 1, + "cost_microusd": 0, + }, + actor="daemon:planner", + ) + journal.append( + run.run_id, + EXECUTION_PLAN_ADMITTED, + { + "plan_id": substituted_plan.plan_id, + "plan_digest": substituted_plan.plan_digest, + "plan_version": substituted_plan.plan_revision, + "supersedes_plan_id": substituted_plan.supersedes_plan_id, + "plan": plan_payload(substituted_plan), + }, + actor="daemon:planner", + ) + + with pytest.raises(RuntimeApiError) as caught: + runtime.inspect_run(run.run_id) + assert caught.value.code is RuntimeApiFailureCode.CONFLICT + + +@pytest.mark.parametrize("after_draft", [False, True]) +def test_generated_runtime_recovers_interrupted_initial_planning( + tmp_path: Path, + *, + after_draft: bool, +) -> None: + repository = _repository(tmp_path) + database = tmp_path / f"public-generated-interrupted-{after_draft}.sqlite3" + events = EventStore(database) + runtime = RuntimeService(events, repository) + intent = msgspec.structs.replace(_intent(), unresolved_questions=()) + bounds = msgspec.structs.replace(_plan(repository), planning_mode="generated") + run = _run() + runtime.register_project(_project(repository), principal_id="client:test") + runtime.accept_intent(intent, principal_id="client:test") + runtime.accept_plan(bounds, principal_id="client:test") + runtime.submit_run(run, principal_id="client:test") + generated = runtime.next_generated_run() + assert generated is not None + + task = _task("verify", allowed_paths=()) + task["checks"] = ["inspect-pass", "verify-pass"] + draft = _draft(tasks=(task,)) + journal = EventBackedExecutionRunJournal(events, CheckpointStore(database)) + _append_execution_goal( + journal, + run_id=run.run_id, + goal=generated.goal, + budget=generated.authority.budget, + ) + if after_draft: + _append_execution_draft(journal, run_id=run.run_id, draft=draft) + + selected = runtime.next_generated_run() + assert selected is not None + assert selected.run_id == run.run_id + provider = _Provider(draft) + kernel = ProductionExecution( + ExecutionCoordinator( + journal, + provider, + _Executor((_success(),)), + ExecutionPolicyKernel(), + ), + authority_for_run=runtime.generated_execution_authority, + goal_for_run=runtime.generated_execution_goal, + ) + request = PlanningRequest( + goal=selected.goal, + classification=DataClassification.PRIVATE, + locality=LocalityPolicy.REMOTE_ALLOWED, + budget=selected.authority.budget, + estimated_input_tokens=1_000, + correlation_id=run.run_id, + run_id=run.run_id, + ) + + state = kernel.process(request, actor="daemon:worker") + + assert state.status is RunLifecycleStatus.ESCALATED + assert provider.calls == 0 + assert runtime.inspect_run(run.run_id).status == "reconciliation-required" + terminal = journal.events(run.run_id)[-1] + assert terminal.event_type == EXECUTION_RUN_TERMINATED + assert terminal.payload["reason"] == "ambiguous-planning-dispatch" + + +@pytest.mark.parametrize("substitution", ["coverage", "definition"]) +def test_generated_runtime_rejects_substitution_in_any_plan_revision( + tmp_path: Path, + *, + substitution: str, +) -> None: + repository = _repository(tmp_path) + database = tmp_path / f"public-generated-revision-substitution-{substitution}.sqlite3" + events = EventStore(database) + runtime = RuntimeService(events, repository) + intent = msgspec.structs.replace(_intent(), unresolved_questions=()) + bounds = msgspec.structs.replace(_plan(repository), planning_mode="generated") + run = _run() + runtime.register_project(_project(repository), principal_id="client:test") + runtime.accept_intent(intent, principal_id="client:test") + runtime.accept_plan(bounds, principal_id="client:test") + runtime.submit_run(run, principal_id="client:test") + generated = runtime.next_generated_run() + assert generated is not None + + task = _task("verify", allowed_paths=()) + task["checks"] = ["inspect-pass", "verify-pass"] + draft = _draft(tasks=(task,)) + canonical = compile_plan(generated.goal, draft) + checks = canonical.tasks[0].checks + if substitution == "coverage": + substituted_checks = checks[:1] + else: + substituted_checks = ( + replace(checks[0], argv=("python", "-c", "raise SystemExit(0)")), + *checks[1:], + ) + substituted_plan = replace( + canonical, + tasks=(replace(canonical.tasks[0], checks=substituted_checks),), + ) + journal = EventBackedExecutionRunJournal(events, CheckpointStore(database)) + _append_execution_goal( + journal, + run_id=run.run_id, + goal=generated.goal, + budget=generated.authority.budget, + ) + _append_execution_draft(journal, run_id=run.run_id, draft=draft) + _append_execution_plan(journal, run_id=run.run_id, plan=substituted_plan) + failure = AttemptEvidence( + workspace_clean=True, + verifier_exit_code=1, + required_checks_passed=False, + failure_class=FailureClass.INVALID_ASSUMPTION, + failure_summary="the admitted assumption is invalid", + artifact_digests=(DIGEST,), + progress_digests=(DIGEST,), + head_commit=FAILED_HEAD, + input_tokens=0, + output_tokens=0, + latency_ms=1, + cost_microusd=0, + ) + coordinator = ExecutionCoordinator( + journal, + _Provider(draft), + _SequentialExecutor((failure, _success())), + ExecutionPolicyKernel(), + ) + request = PlanningRequest( + goal=generated.goal, + classification=DataClassification.PRIVATE, + locality=LocalityPolicy.REMOTE_ALLOWED, + budget=generated.authority.budget, + estimated_input_tokens=1_000, + correlation_id=run.run_id, + run_id=run.run_id, + ) + + first = coordinator.execute( + run.run_id, + request, + substituted_plan, + actor="daemon:worker", + ) + assert first.status is RunLifecycleStatus.REPLAN_REQUIRED + canonical_successor, _ = coordinator.compile_and_admit( + run.run_id, + request, + previous=substituted_plan, + actor="daemon:planner", + ) + final = coordinator.execute( + run.run_id, + request, + canonical_successor, + actor="daemon:worker", + ) + assert final.status is RunLifecycleStatus.SUCCEEDED + + with pytest.raises(RuntimeApiError) as caught: + runtime.inspect_run(run.run_id) + assert caught.value.code is RuntimeApiFailureCode.CONFLICT + with pytest.raises(RuntimeApiError) as discovery: + runtime.review_run_ids() + assert discovery.value.code is RuntimeApiFailureCode.CONFLICT + + def test_public_cancellation_during_attempt_stops_verification_and_terminates( tmp_path: Path, ) -> None: @@ -740,6 +1100,15 @@ def remove_run_id(payload: dict[str, object]) -> None: def add_unknown_field(payload: dict[str, object]) -> None: payload["unknown"] = True + def remove_retained_workspace_history(payload: dict[str, object]) -> None: + payload.pop("retained_workspace_ids") + + def remove_current_task_state_fields(payload: dict[str, object]) -> None: + tasks = cast("list[object]", payload["tasks"]) + task = cast("dict[str, object]", tasks[0]) + task.pop("retained_workspace_id") + task.pop("blocked_reason") + def partial_pending_policy(payload: dict[str, object]) -> None: tasks = cast("list[object]", payload["tasks"]) cast("dict[str, object]", tasks[0])["pending_policy_decision_id"] = DIGEST @@ -755,6 +1124,8 @@ def misplaced_pending_policy(payload: dict[str, object]) -> None: mutations = ( remove_run_id, add_unknown_field, + remove_retained_workspace_history, + remove_current_task_state_fields, top_level("run_id", ""), top_level("goal_digest", "bad"), top_level("goal_base_commit", "bad"), @@ -773,6 +1144,8 @@ def misplaced_pending_policy(payload: dict[str, object]) -> None: task_level("last_artifact_digests", "not-a-list"), task_level("last_artifact_digests", [DIGEST, DIGEST]), task_level("head_commit", "bad"), + task_level("retained_workspace_id", ""), + task_level("blocked_reason", "unexpected-policy-state"), task_level("pending_policy_allowed", "true"), partial_pending_policy, misplaced_pending_policy, @@ -1415,6 +1788,9 @@ def test_no_progress_breaker_escalates_and_emits_typed_praxis_candidate( artifact_digests=(DIGEST,), progress_digests=(DIGEST,), head_commit=FAILED_HEAD, + input_tokens=0, + output_tokens=0, + cost_microusd=0, ) provider = _Provider(_draft(tasks=(_task("implement", allowed_paths=("src",)),))) executor = _Executor((failure, failure, _success())) @@ -1519,15 +1895,29 @@ def test_replan_creates_one_immutable_successor_then_executes(tmp_path: Path) -> ) executor = _SequentialExecutor((first_failure, _success())) journal = _journal(tmp_path) + durable_budget = GatewayBudget(10, 10, 1_000, 10) + authority_budget = GatewayBudget(100, 100, 10_000, 100) kernel = ProductionExecution( - ExecutionCoordinator(journal, provider, executor, ExecutionPolicyKernel()) + ExecutionCoordinator(journal, provider, executor, ExecutionPolicyKernel()), + authority_for_run=lambda run_id: _journal_authority( + journal, + run_id, + authority_budget, + ), + ) + request = replace( + _planning_request("run-replan"), + budget=durable_budget, ) - request = _planning_request("run-replan") first = kernel.process(request, actor="daemon:worker") first_plan = journal.plan("run-replan") second = kernel.process(request, actor="daemon:worker") second_plan = journal.plan("run-replan") + restarted = EventBackedExecutionRunJournal( + EventStore(tmp_path / "kernel.sqlite3"), + CheckpointStore(tmp_path / "kernel.sqlite3"), + ).rehydrate("run-replan") assert first.status is RunLifecycleStatus.REPLAN_REQUIRED assert second.status is RunLifecycleStatus.SUCCEEDED @@ -1535,25 +1925,32 @@ def test_replan_creates_one_immutable_successor_then_executes(tmp_path: Path) -> assert second_plan.plan_revision == 2 assert second_plan.supersedes_plan_id == first_plan.plan_id assert second_plan.plan_id != first_plan.plan_id + assert first.retained_workspace_ids == (executor.workspaces[0],) + assert second.retained_workspace_ids == first.retained_workspace_ids + assert restarted.retained_workspace_ids == first.retained_workspace_ids assert provider.calls == 2 + assert provider.budgets == [ + durable_budget, + GatewayBudget(8, 8, 998, 10), + ] assert executor.calls == 2 assert kernel.process(request, actor="daemon:worker") == second assert provider.calls == 2 -def test_second_replan_request_exhausts_budget_without_another_provider_call( +def test_direct_process_replan_uses_durable_remainder_without_public_authority( tmp_path: Path, ) -> None: - invalid_assumption = AttemptEvidence( + first_failure = AttemptEvidence( workspace_clean=True, verifier_exit_code=1, required_checks_passed=False, failure_class=FailureClass.INVALID_ASSUMPTION, - failure_summary="the admitted assumption remains invalid", + failure_summary="the admitted assumption is invalid", artifact_digests=(DIGEST,), progress_digests=(DIGEST,), head_commit=FAILED_HEAD, - input_tokens=1, + input_tokens=2, output_tokens=1, latency_ms=1, cost_microusd=0, @@ -1561,39 +1958,228 @@ def test_second_replan_request_exhausts_budget_without_another_provider_call( provider = _SequenceProvider( ( _draft(tasks=(_task("discover", allowed_paths=("src",)),)), - _draft(tasks=(_task("discover", allowed_paths=("src",)),)), + _draft(tasks=(_task("implement", allowed_paths=("src",)),)), ) ) - executor = _SequentialExecutor((invalid_assumption, invalid_assumption)) - journal = _journal(tmp_path) + journal = _named_journal(tmp_path, "direct-process-remainder") kernel = ProductionExecution( - ExecutionCoordinator(journal, provider, executor, ExecutionPolicyKernel()) + ExecutionCoordinator( + journal, + provider, + _SequentialExecutor((first_failure, _success())), + ExecutionPolicyKernel(), + ) + ) + request = replace( + _planning_request("run-direct-process-remainder"), + budget=GatewayBudget(10, 10, 1_000, 10), ) - request = _planning_request("run-replan-exhausted") first = kernel.process(request, actor="daemon:worker") second = kernel.process(request, actor="daemon:worker") - terminal = kernel.process(request, actor="daemon:worker") - terminal_payload = cast( - "dict[str, JsonValue]", - thaw_json(journal.events(request.run_id)[-1].payload), - ) assert first.status is RunLifecycleStatus.REPLAN_REQUIRED - assert second.status is RunLifecycleStatus.REPLAN_REQUIRED - assert journal.plan(request.run_id).plan_revision == 2 - assert terminal.status is RunLifecycleStatus.ESCALATED - assert terminal_payload["reason"] == "replan-budget-exhausted" - assert provider.calls == executor.calls == 2 - assert kernel.process(request, actor="daemon:worker") == terminal - assert provider.calls == executor.calls == 2 + assert second.status is RunLifecycleStatus.SUCCEEDED + assert provider.budgets == [ + request.budget, + GatewayBudget(7, 8, 998, 10), + ] -def test_dependency_heads_cover_diamond_redundant_and_divergent_edges(tmp_path: Path) -> None: - draft = _draft( - tasks=( - _task("write", allowed_paths=("src",)), - _task("left", depends_on=("write",)), +def test_direct_run_replan_uses_durable_remainder_without_public_authority( + tmp_path: Path, +) -> None: + failure = AttemptEvidence( + workspace_clean=True, + verifier_exit_code=1, + required_checks_passed=False, + failure_class=FailureClass.INVALID_ASSUMPTION, + failure_summary="the admitted assumption is invalid", + artifact_digests=(DIGEST,), + progress_digests=(DIGEST,), + head_commit=FAILED_HEAD, + input_tokens=1, + output_tokens=1, + latency_ms=1, + cost_microusd=0, + ) + provider = _SequenceProvider( + ( + _draft(tasks=(_task("discover", allowed_paths=("src",)),)), + _draft(tasks=(_task("implement", allowed_paths=("src",)),)), + ) + ) + kernel = ProductionExecution( + ExecutionCoordinator( + _named_journal(tmp_path, "direct-run-remainder"), + provider, + _SequentialExecutor((failure, _success())), + ExecutionPolicyKernel(), + ) + ) + request = replace( + _planning_request("run-direct-run-remainder"), + budget=GatewayBudget(10, 10, 1_000, 10), + ) + + first = kernel.run(request, actor="daemon:worker") + second = kernel.run(request, actor="daemon:worker", previous=first.plan) + + assert first.state.status is RunLifecycleStatus.REPLAN_REQUIRED + assert second.state.status is RunLifecycleStatus.SUCCEEDED + assert provider.budgets == [ + request.budget, + GatewayBudget(8, 8, 998, 10), + ] + + +@pytest.mark.parametrize("incomplete", ("input", "output", "cost")) +def test_direct_replan_stops_before_provider_when_usage_is_incomplete( + tmp_path: Path, + incomplete: str, +) -> None: + failure = AttemptEvidence( + workspace_clean=True, + verifier_exit_code=1, + required_checks_passed=False, + failure_class=FailureClass.INVALID_ASSUMPTION, + failure_summary="the admitted assumption is invalid", + artifact_digests=(DIGEST,), + progress_digests=(DIGEST,), + head_commit=FAILED_HEAD, + input_tokens=None if incomplete == "input" else 0, + output_tokens=None if incomplete == "output" else 0, + latency_ms=1, + cost_microusd=None if incomplete == "cost" else 0, + ) + provider = _Provider( + _draft(tasks=(_task("discover", allowed_paths=("src",)),)), + input_tokens=0, + output_tokens=0, + cost_microusd=0, + ) + kernel = ProductionExecution( + ExecutionCoordinator( + _named_journal(tmp_path, f"direct-incomplete-{incomplete}"), + provider, + _SequentialExecutor((failure,)), + ExecutionPolicyKernel(), + ) + ) + request = replace( + _planning_request(f"run-direct-incomplete-{incomplete}"), + budget=GatewayBudget(10, 10, 1_000, 10), + ) + + first = kernel.process(request, actor="daemon:worker") + second = kernel.process(request, actor="daemon:worker") + + assert first.status is RunLifecycleStatus.REPLAN_REQUIRED + assert second.status is RunLifecycleStatus.ESCALATED + assert provider.calls == 1 + + +def test_replan_stops_before_provider_when_durable_usage_is_incomplete( + tmp_path: Path, +) -> None: + first_failure = AttemptEvidence( + workspace_clean=True, + verifier_exit_code=1, + required_checks_passed=False, + failure_class=FailureClass.INVALID_ASSUMPTION, + failure_summary="the admitted assumption is invalid", + artifact_digests=(DIGEST,), + progress_digests=(DIGEST,), + head_commit=FAILED_HEAD, + input_tokens=None, + output_tokens=1, + latency_ms=1, + cost_microusd=0, + ) + provider = _Provider(_draft(tasks=(_task("discover", allowed_paths=("src",)),))) + executor = _SequentialExecutor((first_failure,)) + journal = _journal(tmp_path) + total_budget = GatewayBudget(10, 10, 1_000, 10) + kernel = ProductionExecution( + ExecutionCoordinator(journal, provider, executor, ExecutionPolicyKernel()), + authority_for_run=lambda run_id: _journal_authority( + journal, + run_id, + total_budget, + ), + ) + request = replace( + _planning_request("run-incomplete-replan"), + budget=total_budget, + ) + + first = kernel.process(request, actor="daemon:worker") + second = kernel.process(request, actor="daemon:worker") + terminal_payload = cast( + "dict[str, JsonValue]", + thaw_json(journal.events(request.run_id)[-1].payload), + ) + + assert first.status is RunLifecycleStatus.REPLAN_REQUIRED + assert second.status is RunLifecycleStatus.ESCALATED + assert terminal_payload["reason"] == "cumulative-budget-exhausted" + assert provider.calls == 1 + assert executor.calls == 1 + + +def test_second_replan_request_exhausts_budget_without_another_provider_call( + tmp_path: Path, +) -> None: + invalid_assumption = AttemptEvidence( + workspace_clean=True, + verifier_exit_code=1, + required_checks_passed=False, + failure_class=FailureClass.INVALID_ASSUMPTION, + failure_summary="the admitted assumption remains invalid", + artifact_digests=(DIGEST,), + progress_digests=(DIGEST,), + head_commit=FAILED_HEAD, + input_tokens=1, + output_tokens=1, + latency_ms=1, + cost_microusd=0, + ) + provider = _SequenceProvider( + ( + _draft(tasks=(_task("discover", allowed_paths=("src",)),)), + _draft(tasks=(_task("discover", allowed_paths=("src",)),)), + ) + ) + executor = _SequentialExecutor((invalid_assumption, invalid_assumption)) + journal = _journal(tmp_path) + kernel = ProductionExecution( + ExecutionCoordinator(journal, provider, executor, ExecutionPolicyKernel()) + ) + request = _planning_request("run-replan-exhausted") + + first = kernel.process(request, actor="daemon:worker") + second = kernel.process(request, actor="daemon:worker") + terminal = kernel.process(request, actor="daemon:worker") + terminal_payload = cast( + "dict[str, JsonValue]", + thaw_json(journal.events(request.run_id)[-1].payload), + ) + + assert first.status is RunLifecycleStatus.REPLAN_REQUIRED + assert second.status is RunLifecycleStatus.REPLAN_REQUIRED + assert journal.plan(request.run_id).plan_revision == 2 + assert terminal.status is RunLifecycleStatus.ESCALATED + assert terminal_payload["reason"] == "replan-budget-exhausted" + assert provider.calls == executor.calls == 2 + assert kernel.process(request, actor="daemon:worker") == terminal + assert provider.calls == executor.calls == 2 + + +def test_dependency_heads_cover_diamond_redundant_and_divergent_edges(tmp_path: Path) -> None: + draft = _draft( + tasks=( + _task("write", allowed_paths=("src",)), + _task("left", depends_on=("write",)), _task("right", depends_on=("write",)), _task( "merge", @@ -1802,10 +2388,12 @@ def test_event_journal_exports_run_plan_task_and_attempt_correlations(tmp_path: planning = next( item for item in records if item.attributes["event.type"] == EXECUTION_PLAN_DRAFT_RECEIVED ) - assert planning.attributes["usage.input_tokens.known"] is False - assert planning.attributes["usage.output_tokens.known"] is False - assert planning.attributes["usage.cost_microusd.known"] is False - assert "usage.input_tokens" not in planning.attributes + assert planning.attributes["usage.input_tokens.known"] is True + assert planning.attributes["usage.output_tokens.known"] is True + assert planning.attributes["usage.cost_microusd.known"] is True + assert planning.attributes["usage.input_tokens"] == 0 + assert planning.attributes["usage.output_tokens"] == 0 + assert planning.attributes["usage.cost_microusd"] == 0 verified = next( item for item in records if item.attributes["event.type"] == EXECUTION_TASK_VERIFIED ) @@ -1881,6 +2469,27 @@ def test_execution_contracts_reject_malformed_identity_scope_and_evidence() -> N with pytest.raises(PlanContractError): construct() + invalid_authorities = ( + lambda: ExecutionAuthority(cast("GatewayBudget", object()), 30, 1), + lambda: ExecutionAuthority(GatewayBudget(1, 1, 1, 0), 0, 1), + lambda: ExecutionAuthority(GatewayBudget(1, 1, 1, 0), 30, -1), + lambda: ExecutionAuthority( + GatewayBudget(1, 1, 1, 0), + 30, + 1, + GatewayBudget(2, 1, 1, 0), + ), + lambda: ExecutionAuthority( + GatewayBudget(1, 1, 1, 0), + 30, + 1, + input_tokens_complete=cast("bool", 1), + ), + ) + for construct in invalid_authorities: + with pytest.raises(PlanContractError): + construct() + invalid_checks = ( ("empty argv", lambda: VerificationCheck("check", ())), ("unsafe executable", lambda: VerificationCheck("check", ("../pytest",))), @@ -2122,15 +2731,190 @@ def invoke(self, request: ModelRequest) -> GatewayResult: ) +def test_runtime_resumes_legacy_durable_policy_fence_without_reauthorizing( + tmp_path: Path, +) -> None: + request = _planning_request("run-policy-resume") + journal = _named_journal(tmp_path, "policy-resume") + provider = _Provider( + _draft(tasks=(_task("implement", allowed_paths=("src",)),)), + input_tokens=1, + output_tokens=1, + cost_microusd=0, + ) + executor = _Executor((_success(),)) + admitting = ExecutionCoordinator(journal, provider, executor, ExecutionPolicyKernel()) + plan, _ = admitting.compile_and_admit("run-policy-resume", request, actor="daemon:planner") + journal.append( + "run-policy-resume", + EXECUTION_TASK_READY, + {"task_id": "implement", "attempt": 1}, + actor="daemon:worker", + ) + task = plan.tasks[0] + action = ToolActionRequest( + run_id="run-policy-resume", + plan_id=plan.plan_id, + task_id=task.task_id, + attempt=1, + capability="repository-task", + allowed_paths=task.allowed_paths, + ) + decision = ExecutionPolicyKernel().authorize(request.goal, plan, task, action) + journal.append( + "run-policy-resume", + EXECUTION_POLICY_DECIDED, + { + "task_id": task.task_id, + "attempt": 1, + "allowed": decision.allowed, + "reason": decision.reason, + "action_digest": decision.action_digest, + "decision_id": decision.decision_id, + }, + actor="daemon:worker", + ) + resumed = ExecutionCoordinator(journal, provider, executor, _NeverAuthorizePolicy()) + narrower_budget = GatewayBudget(1_000, 1_000, 1_000, 0) + kernel = ProductionExecution( + resumed, + authority_for_run=lambda _: ExecutionAuthority(narrower_budget, 1, 1), + ) + + state = kernel.process( + replace(request, budget=narrower_budget), + actor="daemon:worker", + ) + + assert state.status is RunLifecycleStatus.SUCCEEDED + assert state.budget == request.budget + assert executor.decisions == [decision] + assert ( + sum( + event.event_type == EXECUTION_POLICY_DECIDED + for event in journal.events("run-policy-resume") + ) + == 1 + ) + + +def test_runtime_completes_denied_task_termination_after_blocked_event_restart( + tmp_path: Path, +) -> None: + request = _planning_request("run-policy-blocked-resume") + journal = _named_journal(tmp_path, "policy-blocked-resume") + policy = _DenyPolicy() + coordinator = ExecutionCoordinator( + journal, + _Provider(_draft(tasks=(_task("implement", allowed_paths=("src",)),))), + _Executor((_success(),)), + policy, + ) + plan, _ = coordinator.compile_and_admit( + request.run_id, + request, + actor="daemon:planner", + ) + journal.append( + request.run_id, + EXECUTION_TASK_READY, + {"task_id": "implement", "attempt": 1}, + actor="daemon:worker", + ) + task = plan.tasks[0] + action = ToolActionRequest( + run_id=request.run_id, + plan_id=plan.plan_id, + task_id=task.task_id, + attempt=1, + capability="repository-task", + allowed_paths=task.allowed_paths, + ) + decision = policy.authorize(request.goal, plan, task, action) + journal.append( + request.run_id, + EXECUTION_POLICY_DECIDED, + { + "task_id": task.task_id, + "attempt": 1, + "allowed": decision.allowed, + "reason": decision.reason, + "action_digest": decision.action_digest, + "decision_id": decision.decision_id, + }, + actor="daemon:worker", + ) + journal.append( + request.run_id, + EXECUTION_TASK_BLOCKED, + {"task_id": task.task_id, "reason": decision.reason}, + actor="daemon:worker", + ) + projection = ExecutionRunProjection() + current = journal.rehydrate(request.run_id) + old_payload = deepcopy(cast("dict[str, object]", projection.dump_state(current))) + old_payload.pop("retained_workspace_ids") + old_tasks = cast("list[object]", old_payload["tasks"]) + for old_task in old_tasks: + old_state = cast("dict[str, object]", old_task) + old_state.pop("retained_workspace_id") + old_state.pop("blocked_reason") + blocked_event = journal.events(request.run_id)[-1] + assert blocked_event.global_position is not None + checkpoints = CheckpointStore(tmp_path / "policy-blocked-resume.sqlite3") + checkpoints.save( + ProjectionCheckpoint.create( + projection_name=projection.name, + projection_version=5, + stream_id=f"run:{request.run_id}", + last_global_position=blocked_event.global_position, + last_stream_sequence=blocked_event.stream_sequence, + state=cast("dict[str, JsonValue]", old_payload), + ) + ) + assert checkpoints.load(projection.name, 5, stream_id=f"run:{request.run_id}") is not None + assert ( + checkpoints.load(projection.name, projection.version, stream_id=f"run:{request.run_id}") + is None + ) + stranded = journal.rehydrate(request.run_id) + resumed = ExecutionCoordinator( + journal, + _Provider(_draft(tasks=(_task("implement", allowed_paths=("src",)),))), + _Executor((_success(),)), + _NeverAuthorizePolicy(), + ) + + state = resumed.execute(request.run_id, request, plan, actor="daemon:worker") + + assert stranded.status is RunLifecycleStatus.RUNNING + assert stranded.task(task.task_id).blocked_reason == decision.reason + assert state.status is RunLifecycleStatus.BLOCKED + assert state.task(task.task_id).status is TaskLifecycleStatus.BLOCKED + assert ( + sum( + event.event_type == EXECUTION_POLICY_DECIDED for event in journal.events(request.run_id) + ) + == 1 + ) + terminal = journal.events(request.run_id)[-1] + assert terminal.event_type == EXECUTION_RUN_TERMINATED + terminal_payload = cast("dict[str, JsonValue]", thaw_json(terminal.payload)) + assert terminal_payload["reason"] == decision.reason + + @dataclass class _Provider: draft: dict[str, object] - input_tokens: int | None = None - output_tokens: int | None = None - cost_microusd: int | None = None + input_tokens: int | None = 0 + output_tokens: int | None = 0 + cost_microusd: int | None = 0 + calls: int = 0 + budgets: list[GatewayBudget] = field(default_factory=list) def propose_plan(self, request: PlanningRequest) -> PlanningResult: - del request + self.calls += 1 + self.budgets.append(request.budget) return PlanningResult( draft=cast("dict[str, JsonValue]", self.draft), provider_output_digest=json_digest(cast("dict[str, JsonValue]", self.draft)), @@ -2148,9 +2932,10 @@ def propose_plan(self, request: PlanningRequest) -> PlanningResult: class _SequenceProvider: drafts: tuple[dict[str, object], ...] calls: int = 0 + budgets: list[GatewayBudget] = field(default_factory=list) def propose_plan(self, request: PlanningRequest) -> PlanningResult: - del request + self.budgets.append(request.budget) draft = self.drafts[self.calls] self.calls += 1 return PlanningResult( @@ -2201,6 +2986,7 @@ def execute( class _SequentialExecutor: outcomes: tuple[AttemptEvidence, ...] calls: int = 0 + workspaces: list[str] = field(default_factory=list) def execute( self, @@ -2217,13 +3003,13 @@ def execute( policy_decision: PolicyDecision, remaining_budget: GatewayBudget, ) -> AttemptEvidence: + self.workspaces.append(workspace_id) del ( run_id, goal, plan, task, attempt, - workspace_id, base_commit, prior_failure_class, prior_failure_summary, @@ -2324,6 +3110,18 @@ def authorize( return PolicyDecision(False, "test-policy-denial", request.digest) +class _NeverAuthorizePolicy(ExecutionPolicyKernel): + def authorize( + self, + goal: GoalSpec, + plan: Plan, + task: TaskSpec, + request: ToolActionRequest, + ) -> PolicyDecision: + del goal, plan, task, request + raise AssertionError("a durable policy fence must not be reauthorized") + + class _RaisingExecutor: def execute( self, @@ -2366,6 +3164,33 @@ def _named_journal(tmp_path: Path, name: str) -> EventBackedExecutionRunJournal: return EventBackedExecutionRunJournal(EventStore(path), CheckpointStore(path)) +def _journal_authority( + journal: EventBackedExecutionRunJournal, + run_id: str, + total: GatewayBudget, +) -> ExecutionAuthority: + try: + state = journal.rehydrate(run_id) + except ExecutionRuntimeError as error: + if error.code != "execution-run-not-found": + raise + return ExecutionAuthority(total, 30, 1) + return ExecutionAuthority( + budget=total, + check_timeout_seconds=30, + max_changed_paths=1, + consumed_budget=GatewayBudget( + (state.input_tokens if state.input_tokens_complete else total.max_input_tokens), + (state.output_tokens if state.output_tokens_complete else total.max_output_tokens), + state.latency_ms, + (state.cost_microusd if state.cost_microusd_complete else total.max_cost_microusd), + ), + input_tokens_complete=state.input_tokens_complete, + output_tokens_complete=state.output_tokens_complete, + cost_microusd_complete=state.cost_microusd_complete, + ) + + def _planning_request(run_id: str) -> PlanningRequest: return PlanningRequest( goal=_goal(), @@ -2378,6 +3203,78 @@ def _planning_request(run_id: str) -> PlanningRequest: ) +def _append_execution_goal( + journal: EventBackedExecutionRunJournal, + *, + run_id: str, + goal: GoalSpec, + budget: GatewayBudget, +) -> None: + journal.append( + run_id, + EXECUTION_GOAL_ADMITTED, + { + "goal_id": goal.goal_id, + "goal_digest": goal.digest, + "goal": goal_payload(goal), + "classification": DataClassification.PRIVATE.value, + "locality": LocalityPolicy.REMOTE_ALLOWED.value, + "budget": { + "max_input_tokens": budget.max_input_tokens, + "max_output_tokens": budget.max_output_tokens, + "max_latency_ms": budget.max_latency_ms, + "max_cost_microusd": budget.max_cost_microusd, + }, + }, + actor="daemon:planner", + ) + + +def _append_execution_draft( + journal: EventBackedExecutionRunJournal, + *, + run_id: str, + draft: dict[str, object], +) -> None: + draft_digest = json_digest(cast("dict[str, JsonValue]", draft)) + journal.append( + run_id, + EXECUTION_PLAN_DRAFT_RECEIVED, + { + "draft_digest": draft_digest, + "provider_output_digest": draft_digest, + "profile_id": "recorded-plan", + "adapter_id": "recorded-plan", + "model_id": "recorded-plan", + "input_tokens": 1, + "output_tokens": 1, + "latency_ms": 1, + "cost_microusd": 0, + }, + actor="daemon:planner", + ) + + +def _append_execution_plan( + journal: EventBackedExecutionRunJournal, + *, + run_id: str, + plan: Plan, +) -> None: + journal.append( + run_id, + EXECUTION_PLAN_ADMITTED, + { + "plan_id": plan.plan_id, + "plan_digest": plan.plan_digest, + "plan_version": plan.plan_revision, + "supersedes_plan_id": plan.supersedes_plan_id, + "plan": plan_payload(plan), + }, + actor="daemon:planner", + ) + + def _goal(*, allowed_paths: tuple[str, ...] = ("src",)) -> GoalSpec: return GoalSpec( goal_id="goal-execution", @@ -2420,4 +3317,7 @@ def _success() -> AttemptEvidence: artifact_digests=(DIGEST,), progress_digests=(DIGEST,), head_commit=SUCCESS_HEAD, + input_tokens=0, + output_tokens=0, + cost_microusd=0, ) diff --git a/tests/unit/test_runtime_service.py b/tests/unit/test_runtime_service.py index 6aab352..fdb88cd 100644 --- a/tests/unit/test_runtime_service.py +++ b/tests/unit/test_runtime_service.py @@ -13,6 +13,7 @@ WorktreeLifecycleError, ) from blackcell.bootstrap.runtime_service import RuntimeService +from blackcell.gateway import DataClassification, GatewayBudget, LocalityPolicy from blackcell.interfaces.http import ( AcceptanceCheck, CancelRunRequest, @@ -26,7 +27,18 @@ RuntimeApiError, RuntimeApiFailureCode, ) -from blackcell.kernel import ArtifactStore, EventEnvelope, EventStore +from blackcell.kernel import ArtifactStore, CheckpointStore, EventEnvelope, EventStore, JsonValue +from blackcell.orchestration.execution_plan import ( + EXECUTION_PLAN_DRAFT_SCHEMA, + ExecutionPolicyKernel, + PlanningRequest, + PlanningResult, + TaskAttemptExecutor, +) +from blackcell.orchestration.execution_runtime import ( + EventBackedExecutionRunJournal, + ExecutionCoordinator, +) _CONFIGURATION_DIGEST = "sha256:" + ("a" * 64) _OTHER_CONFIGURATION_DIGEST = "sha256:" + ("b" * 64) @@ -44,6 +56,34 @@ def retain_plan_base_commit( raise WorktreeLifecycleError(WorktreeFailureCode.BASE_COMMIT_RETENTION_FAILED) +class UnknownUsagePlanner: + def propose_plan(self, request: PlanningRequest) -> PlanningResult: + del request + draft = { + "schema_version": EXECUTION_PLAN_DRAFT_SCHEMA, + "tasks": [ + { + "task_id": "verify", + "objective": "Run the admitted verification checks.", + "depends_on": [], + "allowed_paths": [], + "checks": ["inspect-pass", "verify-pass"], + } + ], + } + return PlanningResult( + draft=cast("dict[str, JsonValue]", draft), + provider_output_digest=_CONFIGURATION_DIGEST, + profile_id="unknown-usage-planner", + adapter_id="recorded-test", + model_id="recorded-test", + input_tokens=None, + output_tokens=None, + latency_ms=7, + cost_microusd=None, + ) + + def test_runtime_flow_is_idempotent_restart_safe_and_live_free(tmp_path: Path) -> None: repository = _repository(tmp_path) events = EventStore(tmp_path / "data" / "blackcell.sqlite3") @@ -83,7 +123,8 @@ def test_generated_plan_run_uses_asynchronous_execution_selection( tmp_path: Path, ) -> None: repository = _repository(tmp_path) - service = RuntimeService(EventStore(tmp_path / "generated.sqlite3"), repository) + events = EventStore(tmp_path / "generated.sqlite3") + service = RuntimeService(events, repository) service.register_project(_project(repository), principal_id="operator") intent = IntentRequest( schema_version="intent-request/v1", @@ -104,7 +145,18 @@ def test_generated_plan_run_uses_asynchronous_execution_selection( intent_id=declared.intent_id, base_commit=declared.base_commit, allowed_effects=declared.allowed_effects, - nodes=declared.nodes, + nodes=tuple( + PlanNode( + node_id=node.node_id, + objective=node.objective, + depends_on=node.depends_on, + budget=NodeBudget(1_000, 1_000, 30, 50, 0), + effects=node.effects, + allowed_paths=node.allowed_paths, + checks=node.checks, + ) + for node in declared.nodes + ), idempotency_key=declared.idempotency_key, planning_mode="generated", ) @@ -118,11 +170,45 @@ def test_generated_plan_run_uses_asynchronous_execution_selection( assert selected.run_id == "run-1" assert selected.goal.objective == intent.objective assert selected.goal.base_commit == generated.base_commit + assert selected.authority.check_timeout_seconds == 30 + assert selected.authority.max_changed_paths == 0 + assert selected.authority.budget.max_input_tokens == 2_000 + assert selected.authority.budget.max_output_tokens == 2_000 + assert selected.authority.budget.max_latency_ms == 60_000 + assert selected.authority.budget.max_cost_microusd == 100 + assert service.generated_execution_authority(selected.run_id) == selected.authority assert tuple(item.check_id for item in selected.goal.verification_checks) == ( "inspect-pass", "verify-pass", ) + coordinator = ExecutionCoordinator( + EventBackedExecutionRunJournal(events, CheckpointStore(events.path)), + UnknownUsagePlanner(), + cast("TaskAttemptExecutor", object()), + ExecutionPolicyKernel(), + ) + coordinator.compile_and_admit( + selected.run_id, + PlanningRequest( + goal=selected.goal, + classification=DataClassification.PRIVATE, + locality=LocalityPolicy.REMOTE_ALLOWED, + budget=selected.authority.budget, + estimated_input_tokens=1, + correlation_id=selected.run_id, + run_id=selected.run_id, + ), + actor="test:planner", + ) + + resumed = service.generated_execution_authority(selected.run_id) + + assert resumed.consumed_budget == GatewayBudget(2_000, 2_000, 7, 100) + assert not resumed.input_tokens_complete + assert not resumed.output_tokens_complete + assert not resumed.cost_microusd_complete + def test_runtime_submission_rejects_mismatched_references_and_conflicts(tmp_path: Path) -> None: repository = _repository(tmp_path)