[代码熵回收] 批次仓储承担业务计数与任务用途判定

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))

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}

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"]):

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

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