diff --git a/src/chatcopilot/harness/governance_run_service.py b/src/chatcopilot/harness/governance_run_service.py index 8c6d8fe..e4cac73 100644 --- a/src/chatcopilot/harness/governance_run_service.py +++ b/src/chatcopilot/harness/governance_run_service.py @@ -8,12 +8,12 @@ from chatcopilot.harness.config import safe_error from chatcopilot.harness.control_types import WorkerState from chatcopilot.harness.governance_run_repository import GovernanceRunRepository -from chatcopilot.harness.governance_types import GovernanceOptions, GovernanceTaskPort, RUN_ACTIVE +from chatcopilot.harness.governance_types import GovernanceLifecyclePort, GovernanceOptions, GovernanceTaskPort, RUN_ACTIVE from chatcopilot.harness.models import ACTIVE, HarnessError, RepairOptions class GovernanceRuns: - def __init__(self, store, lifecycle, tasks: GovernanceTaskPort, repository: str): + def __init__(self, store, lifecycle: GovernanceLifecyclePort, tasks: GovernanceTaskPort, repository: str): self.store, self.lifecycle, self.tasks, self.repository = store, lifecycle, tasks, repository self.runs = GovernanceRunRepository(store) diff --git a/src/chatcopilot/harness/governance_types.py b/src/chatcopilot/harness/governance_types.py index 6c81844..84cc4de 100644 --- a/src/chatcopilot/harness/governance_types.py +++ b/src/chatcopilot/harness/governance_types.py @@ -4,6 +4,7 @@ from dataclasses import asdict, dataclass, field from typing import Any, Literal, Protocol, TypedDict +from chatcopilot.harness.control_types import WorkerState from chatcopilot.harness.models import RepairOptions @@ -78,3 +79,16 @@ def start_learning(self, run: dict[str, Any], sequence: int, options: RepairOpti def launch(self, task_id: str) -> None: ... def resume(self, task_id: str) -> None: ... def retry_delivery(self, task_id: str) -> None: ... + + +class GovernanceWorkerObservationPort(Protocol): + def observe(self, task: dict[str, Any]) -> WorkerState: ... + + +class GovernanceLifecyclePort(Protocol): + @property + def workers(self) -> GovernanceWorkerObservationPort: ... + + def cancel(self, task_id: str) -> dict[str, Any]: ... + def reconcile(self, task_id: str) -> dict[str, Any]: ... + def require_stopped(self, task: dict[str, Any]) -> None: ... diff --git a/tests/unit/harness_regressions/test_292b37061f2f3c65ae86d84b80166feddd4d2d83c5e66dbad20632f008fbd93c.py b/tests/unit/harness_regressions/test_292b37061f2f3c65ae86d84b80166feddd4d2d83c5e66dbad20632f008fbd93c.py new file mode 100644 index 0000000..751b5cc --- /dev/null +++ b/tests/unit/harness_regressions/test_292b37061f2f3c65ae86d84b80166feddd4d2d83c5e66dbad20632f008fbd93c.py @@ -0,0 +1,196 @@ +"""GC lifecycle regression through the controller's current batch assembly.""" + +from dataclasses import asdict +from types import SimpleNamespace +from unittest.mock import Mock + +import pytest + +from chatcopilot.harness.api import HarnessController +from chatcopilot.harness.control_types import DispatchResult, WorkerState +from chatcopilot.harness.governance_types import GovernanceOptions +from chatcopilot.harness.models import HarnessError + + +class ObservedWorkers: + def __init__(self): + self.states = {} + self.launches = [] + self.delivery_launches = [] + + def observe(self, task): + return self.states.get(task["task_id"], WorkerState.INACTIVE) + + def launch(self, task): + self.launches.append(task["task_id"]) + self.states[task["task_id"]] = WorkerState.ACTIVE + return DispatchResult("scheduled") + + def launch_delivery(self, task): + self.delivery_launches.append(task["task_id"]) + return DispatchResult("scheduled") + + +class RecordedTaskOperations: + """The Service's task execution port; persistence and lifecycle stay real.""" + + def __init__(self, store, lifecycle): + self.store = store + self.lifecycle = lifecycle + self.started = [] + self.lose_receipt = False + + def preflight_model(self, model, reasoning_effort): + assert (model, reasoning_effort) == ("fixture", "medium") + + def start(self, run, sequence, options): + key = f"{run['run_id']}-{sequence}" + task, created = self.store.create({ + "task_id": "repair-" + key, + "pipeline_version": 10, + "request_key": key, + "request_digest": key, + "context_key": "repository", + "match_key": key, + "active_key": key, + "source": {"kind": "code_health"}, + "options": asdict(options), + "repository": run["repository"], + "base_commit": "fixture-main", + "unit": key, + "governance_run_id": run["run_id"], + "governance_sequence": sequence, + "dispatch_state": "creating", + "delivery": {"state": "pending"}, + }) + if created: + self.started.append(task["task_id"]) + if self.lose_receipt: + self.lose_receipt = False + raise RuntimeError("creation receipt lost") + return task + + def start_learning(self, run, sequence, options, source): + raise AssertionError("fixture has no merged Skill lesson") + + def launch(self, task_id): + with self.lifecycle.operation(task_id): + self.lifecycle.launch_locked(task_id) + + def resume(self, task_id): + with self.lifecycle.operation(task_id): + self.lifecycle.prepare_resume_locked(task_id) + self.lifecycle.claim_resume_locked(task_id) + self.lifecycle.launch_locked(task_id) + + def retry_delivery(self, task_id): + self.store.update(task_id, delivery_cancel_requested=False) + + +@pytest.fixture +def batch(tmp_path): + repository = tmp_path / "repo" + repository.mkdir() + workers = ObservedWorkers() + controller = HarnessController( + repository, + root=tmp_path / "private", + evaluator=Mock(), + worker_control=workers, + ) + service = controller._governance_runs() + operations = RecordedTaskOperations(controller.store, controller.lifecycle) + service.tasks = operations + return SimpleNamespace(controller=controller, service=service, store=controller.store, + workers=workers, operations=operations) + + +def options(count=2): + return GovernanceOptions("fixture", stop_condition={"mode": "findings", "count": count}) + + +def test_runtime_assembly_injects_existing_lifecycle(batch): + service = batch.controller._governance_runs() + assert service.lifecycle is batch.controller.lifecycle + assert service.lifecycle.workers is batch.workers + assert service.tasks.controller is batch.controller + assert service.store is batch.controller.store + + +def test_cancel_waits_for_worker_and_delivery_then_resume_requires_stop(batch): + run = batch.service.start(options(), request_id="cancel-run") + task_id = run["current_task_id"] + batch.store.update(task_id, status="fixed", stage="done", delivery={"state": "waiting_checks"}) + + cancelling = batch.service.cancel(run["run_id"]) + assert cancelling["status"] == "cancel_requested" + assert batch.store.get(task_id)["delivery_cancel_requested"] is True + batch.workers.states[task_id] = WorkerState.INACTIVE + assert batch.service.advance(run["run_id"])["status"] == "cancel_requested" + assert batch.workers.delivery_launches + batch.store.update(task_id, delivery={"state": "cancelled"}) + cancelled = batch.service.advance(run["run_id"]) + assert cancelled["status"] == "cancelled" + assert cancelled["stop_reason"] == "cancelled" + assert len(batch.operations.started) == 1 + + batch.workers.states[task_id] = WorkerState.UNKNOWN + with pytest.raises(HarnessError) as error: + batch.service.resume(run["run_id"]) + assert error.value.code == "worker_unavailable" + assert batch.service.get(run["run_id"])["status"] == "cancelled" + + +def test_lost_receipt_and_repeated_callbacks_keep_one_child_and_persist_counts(batch): + batch.operations.lose_receipt = True + run = batch.service.start(options(), request_id="lost-receipt") + task_id = run["current_task_id"] + assert run["status"] == "blocked" + assert run["sequence"] == 1 + assert len(batch.operations.started) == 1 + + restarted_service = batch.controller._governance_runs() + restarted_service.tasks = batch.operations + resumed = restarted_service.resume(run["run_id"]) + assert resumed["current_task_id"] == task_id + assert len(batch.workers.launches) == 1 + for _ in range(2): + assert restarted_service.advance(run["run_id"])["current_task_id"] == task_id + assert len(batch.operations.started) == len(batch.workers.launches) == 1 + + batch.workers.states[task_id] = WorkerState.INACTIVE + batch.store.update(task_id, status="fixed", stage="done", governance_finding_id="finding-one", + accepted_candidate={"attempt": 1}, elapsed_seconds=12, + delivery={"state": "checks_pending"}) + waiting = restarted_service.advance(run["run_id"]) + assert waiting["status"] == "waiting_delivery" + assert (waiting["found_count"], waiting["merged_count"], waiting["elapsed_seconds"]) == (1, 0, 12) + assert waiting["sequence"] == 1 + assert len(batch.operations.started) == 1 + assert restarted_service.runs.get(run["run_id"])["found_count"] == 1 + + batch.store.update(task_id, delivery={"state": "merged"}) + batch.workers.states[task_id] = WorkerState.ACTIVE + assert restarted_service.advance(run["run_id"])["sequence"] == 1 + assert len(batch.operations.started) == 1 + batch.workers.states[task_id] = WorkerState.INACTIVE + next_run = restarted_service.advance(run["run_id"]) + assert next_run["sequence"] == 2 + assert next_run["current_task_id"] != task_id + assert (next_run["found_count"], next_run["merged_count"], next_run["elapsed_seconds"]) == (1, 1, 12) + assert len(batch.operations.started) == 2 + assert len(restarted_service.runs.tasks(run["run_id"])) == 2 + persisted = batch.controller._governance_runs().get(run["run_id"]) + assert (persisted["sequence"], persisted["found_count"], persisted["merged_count"], + persisted["elapsed_seconds"]) == (2, 1, 1, 12) + + second_id = next_run["current_task_id"] + batch.workers.states[second_id] = WorkerState.INACTIVE + batch.store.update(second_id, status="fixed", stage="done", governance_finding_id="finding-two", + accepted_candidate={"attempt": 1}, elapsed_seconds=5, + delivery={"state": "merged"}) + completed = restarted_service.advance(run["run_id"]) + assert (completed["status"], completed["stop_reason"]) == ("completed", "findings_limit") + assert (completed["sequence"], completed["found_count"], completed["merged_count"], + completed["elapsed_seconds"]) == (2, 2, 2, 17) + assert len(batch.operations.started) == 2