diff --git a/src/chatcopilot/harness/api.py b/src/chatcopilot/harness/api.py index 24b4cbf..63f3a3a 100644 --- a/src/chatcopilot/harness/api.py +++ b/src/chatcopilot/harness/api.py @@ -88,7 +88,7 @@ def governance_run(self, run_id): return self._governance_runs().get(run_id) def governance_runs(self, **kwargs): - return self._governance_runs().runs.page(repository=str(self.repository), **kwargs) + return self._governance_runs().page(**kwargs) def active_governance_run(self): runs = self._governance_runs().runs.active(str(self.repository)) diff --git a/src/chatcopilot/harness/governance_run_repository.py b/src/chatcopilot/harness/governance_run_repository.py index bc8cde0..4d5c455 100644 --- a/src/chatcopilot/harness/governance_run_repository.py +++ b/src/chatcopilot/harness/governance_run_repository.py @@ -8,7 +8,7 @@ import uuid from chatcopilot.core.private_sqlite import json_text, private_lock -from chatcopilot.harness.governance_types import RUN_ACTIVE, RUN_ACTIVE_STATUSES +from chatcopilot.harness.governance_types import RUN_ACTIVE_STATUSES from chatcopilot.harness.models import HarnessError @@ -92,29 +92,7 @@ def tasks(self, run_id): "ORDER BY json_extract(payload,'$.governance_sequence')", (run_id,)).fetchall() return [json.loads(row[0]) for row in rows] - def project(self, run): - tasks = self.tasks(run["run_id"]) - values = {"found_count": sum(bool(task.get("governance_finding_id")) and not task.get("skill_learning_origin") - for task in tasks), - "merged_count": sum(task.get("delivery", {}).get("state") == "merged" and not task.get("skill_learning_origin") - for task in tasks), - "elapsed_seconds": sum(float(task.get("elapsed_seconds", 0)) for task in tasks), - "current_task_id": tasks[-1]["task_id"] if tasks else None, "sequence": len(tasks)} - return {**run, **values, "tasks": [{ - **{key: task[key] for key in ( - "task_id", "status", "stage", "base_commit", "governance_sequence", "governance_summary", "governance_finding_id", - "message", "stop_reason", "elapsed_seconds", "delivery") if key in task}, - "purpose": "skill_learning" if task.get("skill_learning_origin") else "code_health", - } for task in tasks]} - - def refresh(self, run_id): - value = self.project(self.get(run_id)) - return self.update(run_id, **{key: value[key] for key in ( - "current_task_id", "sequence", "found_count", "merged_count", "elapsed_seconds")}) - def page(self, *, repository, page=1, limit=20, search="", status=""): - if page < 1 or not 1 <= limit <= 100 or status and status not in RUN_ACTIVE | {"completed", "blocked", "cancelled"}: - raise ValueError("无效的批次分页或状态") clause, params = "repository=? AND run_id LIKE ?", [repository, "%" + search + "%"] if status: clause += " AND status=?" @@ -123,4 +101,4 @@ def page(self, *, repository, page=1, limit=20, search="", status=""): total = connection.execute("SELECT COUNT(*) FROM governance_runs WHERE " + clause, params).fetchone()[0] rows = connection.execute("SELECT payload FROM governance_runs WHERE " + clause + " ORDER BY created_at DESC LIMIT ? OFFSET ?", (*params, limit, (page - 1) * limit)).fetchall() - return {"runs": [self.project(json.loads(row[0])) for row in rows], "total": total} + return {"runs": [json.loads(row[0]) for row in rows], "total": total} diff --git a/src/chatcopilot/harness/governance_run_service.py b/src/chatcopilot/harness/governance_run_service.py index e4cac73..99f9507 100644 --- a/src/chatcopilot/harness/governance_run_service.py +++ b/src/chatcopilot/harness/governance_run_service.py @@ -12,6 +12,22 @@ from chatcopilot.harness.models import ACTIVE, HarnessError, RepairOptions +def project_governance_run(run, tasks): + """Derive the public batch view from its stored record and ordered tasks.""" + values = {"found_count": sum(bool(task.get("governance_finding_id")) and not task.get("skill_learning_origin") + for task in tasks), + "merged_count": sum(task.get("delivery", {}).get("state") == "merged" and not task.get("skill_learning_origin") + for task in tasks), + "elapsed_seconds": sum(float(task.get("elapsed_seconds", 0)) for task in tasks), + "current_task_id": tasks[-1]["task_id"] if tasks else None, "sequence": len(tasks)} + return {**run, **values, "tasks": [{ + **{key: task[key] for key in ( + "task_id", "status", "stage", "base_commit", "governance_sequence", "governance_summary", "governance_finding_id", + "message", "stop_reason", "elapsed_seconds", "delivery") if key in task}, + "purpose": "skill_learning" if task.get("skill_learning_origin") else "code_health", + } for task in tasks]} + + class GovernanceRuns: def __init__(self, store, lifecycle: GovernanceLifecyclePort, tasks: GovernanceTaskPort, repository: str): self.store, self.lifecycle, self.tasks, self.repository = store, lifecycle, tasks, repository @@ -21,7 +37,21 @@ def get(self, run_id): run = self.runs.get(run_id) if run["repository"] != self.repository: raise HarnessError("not_found", "此仓库没有该熵回收批次") - return self.runs.project(run) + return self._project(run) + + def _project(self, run): + return project_governance_run(run, self.runs.tasks(run["run_id"])) + + def _refresh(self, run_id): + value = self._project(self.runs.get(run_id)) + return self.runs.update(run_id, **{key: value[key] for key in ( + "current_task_id", "sequence", "found_count", "merged_count", "elapsed_seconds")}) + + def page(self, *, page=1, limit=20, search="", status=""): + if page < 1 or not 1 <= limit <= 100 or status and status not in RUN_ACTIVE | {"completed", "blocked", "cancelled"}: + raise ValueError("无效的批次分页或状态") + result = self.runs.page(repository=self.repository, page=page, limit=limit, search=search, status=status) + return {"runs": [self._project(run) for run in result["runs"]], "total": result["total"]} def start(self, options: GovernanceOptions, *, request_id=None): request_id = request_id or uuid.uuid4().hex @@ -47,7 +77,7 @@ def remaining(run): return max(0, stop["seconds"] - run["elapsed_seconds"]) if stop["mode"] == "time" else None def _advance(self, run_id, *, preflighted=False): - run = self.runs.refresh(run_id) + run = self._refresh(run_id) if run["status"] not in RUN_ACTIVE: return try: @@ -106,11 +136,11 @@ def _advance(self, run_id, *, preflighted=False): if child.get("governance_run_id") != run_id or child.get("governance_sequence") != run["sequence"] + 1: raise HarnessError("governance_active", "Skill 学习任务不属于当前回收批次") self.store.update(task["task_id"], skill_learning={"state": "queued", "task_id": child["task_id"]}) - self.runs.refresh(run_id) + self._refresh(run_id) self.runs.update(run_id, status="running", stop_reason="", message="核对已合并改善的可复用教训") self.tasks.launch(child["task_id"]) return - run = self.runs.refresh(run_id) + run = self._refresh(run_id) stop = run["options"]["stop_condition"] if stop["mode"] == "findings" and run["found_count"] >= stop["count"]: self.runs.update(run_id, status="completed", stop_reason="findings_limit", message="已完成设定数量的问题及 PR 交付") @@ -130,7 +160,7 @@ def _advance(self, run_id, *, preflighted=False): child = self.tasks.start(run, run["sequence"] + 1, child_options) if child.get("governance_run_id") != run_id or child.get("governance_sequence") != run["sequence"] + 1: raise HarnessError("governance_active", "创建结果不属于当前回收批次,未启动任务") - self.runs.refresh(run_id) + self._refresh(run_id) self.runs.update(run_id, status="running", stop_reason="", message="逐项发现并修复中") self.tasks.launch(child["task_id"]) except Exception as exc: @@ -151,7 +181,7 @@ def cancel(self, run_id): def resume(self, run_id): with self.runs.locked(): self.get(run_id) - run = self.runs.refresh(run_id) + run = self._refresh(run_id) if run["status"] not in {"blocked", "cancelled"}: raise HarnessError("conflict", "只有停止的回收批次可以恢复") if self.runs.active(run["repository"]): diff --git a/src/chatcopilot/harness/task_budget.py b/src/chatcopilot/harness/task_budget.py index 4707bf7..2e4c528 100644 --- a/src/chatcopilot/harness/task_budget.py +++ b/src/chatcopilot/harness/task_budget.py @@ -24,8 +24,9 @@ def __enter__(self): self.runs = None if self.run_id: from chatcopilot.harness.governance_run_repository import GovernanceRunRepository + from chatcopilot.harness.governance_run_service import project_governance_run self.runs = GovernanceRunRepository(self.store) - run = self.runs.project(self.runs.get(self.run_id)) + run = project_governance_run(self.runs.get(self.run_id), self.runs.tasks(self.run_id)) stop = run["options"]["stop_condition"] if stop["mode"] == "time": # Preserve sub-second remainder across child allocations rather diff --git a/tests/unit/harness_regressions/test_fa0cfe3f4f6b946f8992367398438d4cdff2016cc57039e76a7a2b28c935c0ec.py b/tests/unit/harness_regressions/test_fa0cfe3f4f6b946f8992367398438d4cdff2016cc57039e76a7a2b28c935c0ec.py new file mode 100644 index 0000000..0c5da3a --- /dev/null +++ b/tests/unit/harness_regressions/test_fa0cfe3f4f6b946f8992367398438d4cdff2016cc57039e76a7a2b28c935c0ec.py @@ -0,0 +1,192 @@ +"""Behavioral contract for the code-health run projection across public callers.""" + +import json +from types import SimpleNamespace +from unittest.mock import Mock + +import pytest + +from chatcopilot.harness.api import HarnessController +from chatcopilot.harness.control_types import WorkerState +from chatcopilot.harness.governance_run_service import GovernanceRuns +from chatcopilot.harness.governance_types import GovernanceOptions +from chatcopilot.harness.models import HarnessError +from chatcopilot.harness.task_budget import TaskBudget + + +def _controller(tmp_path): + repository = tmp_path / "repository" + repository.mkdir() + return HarnessController( + repository, + root=tmp_path / "private", + evaluator=Mock(), + worker_control=Mock(), + ) + + +def _run(controller, request_id, stop, status="running"): + service = controller._governance_runs() + run, created = service.runs.create( + str(controller.repository), + GovernanceOptions("fixture", stop_condition=stop).to_payload(), + request_id, + ) + assert created + if status != "running": + service.runs.update(run["run_id"], status=status) + return run["run_id"] + + +def _task(controller, run_id, sequence, *, finding=None, delivery="merged", elapsed=0.0, learning=False): + task_id = f"repair-{run_id}-{sequence}" + task, created = controller.store.create({ + "task_id": task_id, + "pipeline_version": 10, + "request_key": task_id, + "request_digest": task_id, + "match_key": task_id, + "context_key": str(controller.repository), + "active_key": task_id, + "source": {"kind": "code_health"}, + "repository": str(controller.repository), + "options": {"timeout_seconds": 3}, + "governance_run_id": run_id, + "governance_sequence": sequence, + "base_commit": f"commit-{sequence}", + "governance_summary": f"finding {sequence}", + **({"skill_learning_origin": "repair-origin"} if learning else {}), + }) + assert created + controller.store.update( + task_id, + status="fixed" if delivery == "merged" else "queued", + stage="done" if delivery == "merged" else "queued", + governance_finding_id=finding, + elapsed_seconds=elapsed, + delivery={"state": delivery}, + message=f"task {sequence}", + stop_reason="", + ) + return task["task_id"] + + +def _stored_run(controller, run_id): + with controller.store.database.connect() as connection: + row = connection.execute( + "SELECT payload FROM governance_runs WHERE run_id=?", (run_id,) + ).fetchone() + return json.loads(row[0]) + + +def test_controller_detail_and_page_preserve_projection_filters_and_storage(tmp_path): + controller = _controller(tmp_path) + target = _run(controller, "first", {"mode": "findings", "count": 2}, status="completed") + first = _task(controller, target, 1, finding="finding-1", elapsed=1.25) + learning = _task(controller, target, 2, finding="learning-finding", elapsed=0.5, learning=True) + latest = _task(controller, target, 3, delivery="pending", elapsed=2.25) + blocked = _run(controller, "second", {"mode": "findings", "count": 2}, status="blocked") + running = _run(controller, "third", {"mode": "findings", "count": 2}) + + stored_before = _stored_run(controller, target) + detail = controller.governance_run(target) + assert (detail["found_count"], detail["merged_count"], detail["elapsed_seconds"]) == (1, 1, 4.0) + assert (detail["current_task_id"], detail["sequence"]) == (latest, 3) + assert [task["task_id"] for task in detail["tasks"]] == [first, learning, latest] + assert [task["purpose"] for task in detail["tasks"]] == ["code_health", "skill_learning", "code_health"] + assert set(detail["tasks"][0]) == { + "task_id", "status", "stage", "base_commit", "governance_sequence", "governance_summary", + "governance_finding_id", "message", "stop_reason", "elapsed_seconds", "delivery", "purpose", + } + assert detail["tasks"][1]["governance_finding_id"] == "learning-finding" + + page = controller.governance_runs(page=1, limit=1) + assert page["total"] == 3 and page["runs"][0]["run_id"] == running + assert controller.governance_runs(page=2, limit=1)["runs"][0]["run_id"] == blocked + completed = controller.governance_runs(status="completed") + assert completed == {"runs": [detail], "total": 1} + assert controller.governance_runs(search=target, status="completed")["total"] == 1 + assert controller.governance_runs(search="absent")["total"] == 0 + with pytest.raises(ValueError): + controller.governance_runs(status="not-a-run-state") + assert _stored_run(controller, target) == stored_before + with controller.store.database.connect() as connection: + assert connection.execute("PRAGMA user_version").fetchone()[0] == 1 + + +def test_task_budget_keeps_fractional_time_across_children(tmp_path): + controller = _controller(tmp_path) + run_id = _run(controller, "budget", {"mode": "time", "seconds": 5}) + _task(controller, run_id, 1, finding="first", elapsed=2.25) + current = _task(controller, run_id, 2, delivery="pending", elapsed=1.25) + clock = [100.0] + + with TaskBudget(controller.store, current, clock=lambda: clock[0]) as budget: + assert budget.remaining == pytest.approx(1.5) + clock[0] += 1.5 + with pytest.raises(HarnessError) as error: + budget.check() + assert error.value.code == "budget_exhausted" + + assert controller.store.get(current)["elapsed_seconds"] == pytest.approx(2.75) + assert controller.governance_run(run_id)["elapsed_seconds"] == pytest.approx(5.0) + + +def test_service_starts_next_item_once_then_stops_at_finding_limit(tmp_path): + controller = _controller(tmp_path) + run_id = _run(controller, "sequence", {"mode": "findings", "count": 2}) + _task(controller, run_id, 1, finding="first", elapsed=2) + lifecycle = SimpleNamespace( + workers=SimpleNamespace(observe=Mock(return_value=WorkerState.INACTIVE)), + reconcile=Mock(), + ) + launched = [] + + def start_next(run, sequence, options): + assert sequence == 2 + assert options.timeout_seconds is None + task_id = _task(controller, run["run_id"], sequence, delivery="pending") + return controller.store.get(task_id) + + port = SimpleNamespace(preflight_model=Mock(), start=Mock(side_effect=start_next), + launch=Mock(side_effect=launched.append)) + service = GovernanceRuns(controller.store, lifecycle, port, str(controller.repository)) + + next_run = service.advance(run_id) + assert next_run["sequence"] == 2 + assert next_run["current_task_id"] == launched[0] + assert port.start.call_count == 1 + controller.store.update(launched[0], status="fixed", governance_finding_id="second", + delivery={"state": "merged"}) + finished = service.advance(run_id) + assert (finished["status"], finished["stop_reason"]) == ("completed", "findings_limit") + assert (finished["found_count"], finished["merged_count"]) == (2, 2) + assert port.start.call_count == len(launched) == 1 + + +def test_service_uses_fractional_remaining_time_and_stops_before_third_item(tmp_path): + controller = _controller(tmp_path) + run_id = _run(controller, "time-sequence", {"mode": "time", "seconds": 5}) + _task(controller, run_id, 1, finding="first", elapsed=2.25) + lifecycle = SimpleNamespace( + workers=SimpleNamespace(observe=Mock(return_value=WorkerState.INACTIVE)), + reconcile=Mock(), + ) + allocated = [] + + def start_next(run, sequence, options): + allocated.append((sequence, options.timeout_seconds)) + task_id = _task(controller, run["run_id"], sequence, delivery="pending") + return controller.store.get(task_id) + + port = SimpleNamespace(preflight_model=Mock(), start=Mock(side_effect=start_next), launch=Mock()) + service = GovernanceRuns(controller.store, lifecycle, port, str(controller.repository)) + + second = service.advance(run_id) + assert allocated == [(2, 3)] + controller.store.update(second["current_task_id"], status="fixed", governance_finding_id="second", + elapsed_seconds=2.75, delivery={"state": "merged"}) + finished = service.advance(run_id) + assert (finished["status"], finished["stop_reason"]) == ("completed", "budget_exhausted") + assert finished["elapsed_seconds"] == pytest.approx(5) + assert port.start.call_count == 1