diff --git a/doc/architecture/SPEC/cli.md b/doc/architecture/SPEC/cli.md index 69abb5aa5..952c14ecf 100644 --- a/doc/architecture/SPEC/cli.md +++ b/doc/architecture/SPEC/cli.md @@ -1,6 +1,6 @@ # cli/ —— 规范 - + `LOC ~1420（6 个文件） · 接口 + 编排门面 · refactor-status: ok` @@ -67,6 +67,8 @@ flag CLI 与 `Copilot` 门面：解析 → 过门 → 执行；并持有 run 目 `resolve()` 与 `_execute` 的 state seed 都走它。 - 阻塞原因随终局落盘：`last_blocked_reason` 进入 `run_status.json` 的 note —— 以前只打在子进程 console 上就被丢掉。 +- resume 成功终局调用 notifier resolution：旧 `ESCALATION.md` 归档到 history， + 不让已经完成的 run 继续显示活动阻塞标记。 ## 边界 —— 不属于这里 不含 step 逻辑、不含仓库知识字面量、不含 LLM prompt。**只做编排接线。** diff --git a/doc/architecture/SPEC/engine/steps/rebase_knowledge.md b/doc/architecture/SPEC/engine/steps/rebase_knowledge.md index 3322140c0..dc480e72c 100644 --- a/doc/architecture/SPEC/engine/steps/rebase_knowledge.md +++ b/doc/architecture/SPEC/engine/steps/rebase_knowledge.md @@ -1,6 +1,6 @@ # engine/steps/rebase_knowledge.py —— 规范 - + `LOC ~354 · step 库（v3 知识尾段） · refactor-status: ok` @@ -24,6 +24,9 @@ v3 playbook 的 Rev 8 §2.2 流水线尾段（`phase5_report → curate → comp report_only 到不了 prep（它的 stores 全程只读）。 - `phase5_report` 在 curate **之前**运行、只读 substate —— curation 永远 没机会掩盖 run 实际做了什么。 +- phase5 事实必须可区分：未分派 module 显示 `skipped (not assigned)`，本地 + 测试分别报告 manifest 总数、实际执行数与 not-run/skipped；远端 CI 同时 + 展示 provider 原始状态和 baseline reconciliation 后的 effective 结果。 - curate 只写 copilot **runtime** store —— 父 read-compat 层与 adapter 树 绝不被写（**D2**）；watchdog 决策收割**恰好一次**（state log 的身份集是 去重权威；`_repo_run_dirs` 按 task.json 定界到本仓库的 run dirs，跨仓库 diff --git a/doc/architecture/SPEC/engine/steps/rebase_v3.md b/doc/architecture/SPEC/engine/steps/rebase_v3.md index 68d8a4a91..b1164da00 100644 --- a/doc/architecture/SPEC/engine/steps/rebase_v3.md +++ b/doc/architecture/SPEC/engine/steps/rebase_v3.md @@ -21,7 +21,7 @@ step —— 薄的受治理 wrapper，substate-first、类型化失败、发布 | `rebase.v3_test_loop` | script/write_workspace | `phase3_failed`；substate `tests.pipeline`/`infra_failures`；空 manifest ⇒ `manifest_empty` | | `rebase.v3_precommit` | script/write_workspace | 无发布 —— substate `tests.precommit`（passed/failed/failed_preexisting/not_declared） | | `rebase.v3_push_gate` | deterministic/read | `push_gate_flagged`、`push_gate_overrides`；不许 ⇒ FORBIDDEN；override 记 trace | -| `rebase.v3_ci` | script/push | `ci_result`、`ci_build_urls`；substate `ci`（rounds/adopted/unfixed，以及逐 job 分类证据） | +| `rebase.v3_ci` | script/push | `ci_result`、`ci_build_urls`；substate `ci`（rounds/adopted/unfixed，以及逐 job 分类证据与 monitor 时长） | | `rebase.v3_finalize` | deterministic/read | 无发布 —— substate `phase=done|needs_human`；有失败 ⇒ BLOCKED（复用 exit 3） | ## 公开契约（注册的 step 之外） diff --git a/doc/architecture/SPEC/engine/steps/report.md b/doc/architecture/SPEC/engine/steps/report.md index f91b790d7..b8d3bb499 100644 --- a/doc/architecture/SPEC/engine/steps/report.md +++ b/doc/architecture/SPEC/engine/steps/report.md @@ -1,18 +1,19 @@ # engine/steps/report.py —— 规范 - + `LOC ~24 · step 库 · refactor-status: ok` ## 职责 -`report.final_summary` —— 用累积的 step 输出写出 `RUN_REPORT.md`。 +`report.final_summary` —— 用累积的 step 输出写出 `RUN_REPORT.md`；repo-rebase +将既有 `FINAL_SUMMARY.md` 嵌入该 canonical 用户报告，不再产出空壳。 ## Steps `report.final_summary`（report/report）。 ## 不变量 - 纯输出；**没有失败路径**（永远 ok）。 -- 读 `ctx.state["outputs"]`；只写 run 目录。 +- 读 `ctx.state["outputs"]` 与同 run 的 rebase summary；只写 run 目录。 ## 边界 —— 不属于这里 不做分析、不含仓库知识、除报告文件外没有任何副作用。 diff --git a/doc/architecture/SPEC/metrics.md b/doc/architecture/SPEC/metrics.md index b50e9cd7f..a91e146d2 100644 --- a/doc/architecture/SPEC/metrics.md +++ b/doc/architecture/SPEC/metrics.md @@ -1,6 +1,6 @@ # metrics.py —— 规范 - + `LOC ~493 · 跨切（测量） · refactor-status: oversized` @@ -8,7 +8,7 @@ 计算并持久化逐 run 的 `metrics.json`（CATQ = Q·S/C）。 ## 功能 -读取该 run 的 trace/产物；计算 Q（按 kind 在**已知**分量上加权的质量）、 +读取该 run 的 trace/产物以及 rebase 的 substate/CI-op/push WAL；计算 Q（按 kind 在**已知**分量上加权的质量）、 S（由 incident 得出的安全系数）、C（RQS3e 式的对数成本，基于 USD + 墙钟对参考预算）； 写出 `metrics.json`。 @@ -24,6 +24,12 @@ S（由 incident 得出的安全系数）、C（RQS3e 式的对数成本，基 - **成本绝不编造。** harness 后端不向 span 记账暴露 token（每项都是 `tok_out=0`）， 所以来源被记为 `subscription`，**不发明 USD** —— 这正是 harness 的成本优势 **是假设而非测量**的原因（`providers/base.md`）。 +- repo-rebase 的 push/build 计数来自 durable WAL，tool 数来自逐次 + `tool_call` 事件；CI monitor wall-time 随 round 持久化并明确标来源。已有 build 但没有可恢复 + 时长、或有 LLM request 却没有 token usage 时，成本标 `partial` 且未知 CI + 分钟写 `null`，绝不以 0 冒充。 +- repo-rebase 的 completed/conflict/tests/purity/push_safe 从 terminal substate、 + WAL 与 incident trace 推导；未观察到的信号继续为 `null`。 ## 边界 —— 不属于这里 **不影响控制流。只做测量。** diff --git a/doc/architecture/SPEC/notify.md b/doc/architecture/SPEC/notify.md index 367fc1923..87bd16d98 100644 --- a/doc/architecture/SPEC/notify.md +++ b/doc/architecture/SPEC/notify.md @@ -1,6 +1,6 @@ # notify.py —— 规范 - + `LOC ~112 · 跨切（升级） · refactor-status: ok` @@ -9,16 +9,18 @@ ## 功能 `Notifier.escalate` 写出 `ESCALATION.md`、发邮件（配置了就走 Resend 或 SMTP）、 -把这次升级记进 trace；`BLOCKED_EXIT` = 3。 +把这次升级记进 trace；恢复后的 run 成功时 `Notifier.resolve` 把活动标记移入 +`ESCALATION_HISTORY.md`；`BLOCKED_EXIT` = 3。 ## 公开契约 `Notifier(settings, run_dir, trace, run_id)`，带 `escalate(reason, phase, -severity, state_summary, artifacts)`；以及 `BLOCKED_EXIT`。 +severity, state_summary, artifacts)`、`resolve(resolution)`；以及 `BLOCKED_EXIT`。 ## 不变量 - 被阻塞的 run 会写出 `ESCALATION.md`、发出通知，调用方以退出码 3 结束。 - **升级是一等结果** —— 绝不被当作错误路径吞掉。 - 邮件失败是尽力而为的，**不得**掩盖升级本身。 +- `ESCALATION.md` 只表示当前仍需关注；成功 resume 后历史保留但活动标记消失。 ## 边界 —— 不属于这里 "要不要升级"由 executor 决定（类型化失败路由）—— 本文件只负责**执行通知**。 diff --git a/doc/architecture/SPEC/rebase_engine.md b/doc/architecture/SPEC/rebase_engine.md index a97018623..c59965e4b 100644 --- a/doc/architecture/SPEC/rebase_engine.md +++ b/doc/architecture/SPEC/rebase_engine.md @@ -87,6 +87,8 @@ runner/LLM/CI client 全部可注入 —— 每个模块都能离线测试。 shell 的 venv 状态绝不影响装到哪；只有 import 验证失败才重试。 - CI 创建受守卫：durable op 记录先于 API 调用，恢复按 op id 精确匹配， **不确定时绝不重复创建**。 +- 每个被 monitor 的 CI round 持久化实际 monitor 时长，供 run metrics 从 durable + substate 计费；无法恢复的旧/中止 build 时长保持 unknown。 - CI baseline 使用最高可信来源（schedule > API > other）的最近窗口：最新 构建失败立即生效，较旧失败只有在窗口内重复出现才保留；单次陈旧红构建不会 越过更新的绿构建复活。 diff --git a/src/infermatrix_copilot/cli/copilot.py b/src/infermatrix_copilot/cli/copilot.py index 5e6938b75..e411b9526 100644 --- a/src/infermatrix_copilot/cli/copilot.py +++ b/src/infermatrix_copilot/cli/copilot.py @@ -434,6 +434,9 @@ def _execute(self, playbook, spec: TaskSpec, run_dir: Path, *, outcome = asyncio.run( run_guarded(executor.run(playbook, state), run_dir)) + if outcome.status == "done": + notifier.resolve() + if self.settings.metrics_enabled: try: # metrics are facts about the run; never let them break it from ..metrics import collect_run_metrics diff --git a/src/infermatrix_copilot/engine/steps/rebase_knowledge.py b/src/infermatrix_copilot/engine/steps/rebase_knowledge.py index e9a979899..cb3d3f781 100644 --- a/src/infermatrix_copilot/engine/steps/rebase_knowledge.py +++ b/src/infermatrix_copilot/engine/steps/rebase_knowledge.py @@ -129,15 +129,23 @@ async def _v3_phase5_report(ctx: StepContext) -> StepResult: "## Modules"] for name, spec in sorted(mods.items()): spec = spec or {} - lines.append(f"- {name}: {spec.get('status', '?')}" + status = ("skipped (not assigned)" if spec.get("skip") else + spec.get("status") or "not completed") + lines.append(f"- {name}: {status}" + (f" — {spec.get('detail')}" if spec.get("detail") else "")) if not mods: lines.append("- (none)") + passed = int(pipeline.get("passed") or 0) + failed = int(pipeline.get("failed") or 0) + skipped = int(pipeline.get("skipped") or 0) + manifest_jobs = int(ctx.state.get("manifest_jobs") or 0) + total = manifest_jobs or passed + failed + skipped lines += ["", "## Local tests", - f"- passed {pipeline.get('passed', 0)} / failed " - f"{pipeline.get('failed', 0)} / skipped " - f"{pipeline.get('skipped', 0)}"] + f"- manifest jobs: {total}", + (f"- executed: {passed + failed} ({passed} passed, " + f"{failed} failed)"), + f"- not run/skipped: {skipped}"] for t in pipeline.get("failed_tests") or []: lines.append(f" - failed: {t}") for i in tests.get("infra_failures") or []: @@ -147,8 +155,16 @@ async def _v3_phase5_report(ctx: StepContext) -> StepResult: lines.append(f"- precommit: {precommit}") if ci: lines += ["", "## Remote CI", - f"- result: {ci.get('result', '')}" + f"- effective result: {ci.get('result', '')}" + (f" ({ci.get('reason')})" if ci.get("reason") else "")] + for round_ in ci.get("rounds") or []: + build = round_.get("build_id") or "?" + lines.append( + f"- raw provider state: {round_.get('build_state') or 'unknown'} " + f"(build {build}; {round_.get('passed', 0)} passed, " + f"{len(round_.get('failed') or [])} unresolved, " + f"{round_.get('ignored_baseline', 0)} baseline, " + f"{round_.get('ignored', 0)} ignored)") for name in ci.get("unfixed") or []: lines.append(f" - unfixed: {name}") path = ctx.run_dir / "FINAL_SUMMARY.md" diff --git a/src/infermatrix_copilot/engine/steps/rebase_v3.py b/src/infermatrix_copilot/engine/steps/rebase_v3.py index 8dcaae2b5..9a4867694 100644 --- a/src/infermatrix_copilot/engine/steps/rebase_v3.py +++ b/src/infermatrix_copilot/engine/steps/rebase_v3.py @@ -2195,6 +2195,7 @@ async def _abort_ci_cleanup(outcome, _client=client, _ops=ops_dir): "ignored": r.ignored, "ignored_baseline": r.ignored_baseline, "jobs": list(r.jobs), + "monitor_duration_sec": r.monitor_duration_sec, } for r in result.rounds]}}) updates = {"ci_result": result.result, "ci_build_urls": [r.build_url for r in result.rounds diff --git a/src/infermatrix_copilot/engine/steps/report.py b/src/infermatrix_copilot/engine/steps/report.py index c4613bf52..2cde73842 100644 --- a/src/infermatrix_copilot/engine/steps/report.py +++ b/src/infermatrix_copilot/engine/steps/report.py @@ -34,6 +34,20 @@ async def _final_report(ctx: StepContext) -> StepResult: if note: lines += [f"- {note}", ""] outputs_map = ctx.state.get("outputs") or {} + if isinstance(spec, dict) and spec.get("kind") == "repo_rebase": + rebase_summary = ctx.run_dir / "FINAL_SUMMARY.md" + if rebase_summary.is_file(): + try: + summary_lines = rebase_summary.read_text( + encoding="utf-8").splitlines() + if summary_lines and summary_lines[0].startswith( + "# FINAL_SUMMARY"): + summary_lines = summary_lines[1:] + while summary_lines and not summary_lines[0].strip(): + summary_lines.pop(0) + lines += ["## Rebase result", "", *summary_lines, ""] + except OSError: + pass for key in _DELIVERABLE_KEYS: text = ctx.state.get(key) if not text: diff --git a/src/infermatrix_copilot/metrics.py b/src/infermatrix_copilot/metrics.py index 682984f1c..2c437acec 100644 --- a/src/infermatrix_copilot/metrics.py +++ b/src/infermatrix_copilot/metrics.py @@ -207,6 +207,43 @@ def _read_json(path: Path) -> dict: return {} +def _durable_rebase_signals(run_dir: Path, substate: dict) -> dict[str, Any]: + """Counts and timings recoverable from the rebase WAL/checkpoints.""" + pushed = 0 + build_ids: set[str] = set() + for path in (run_dir / "push_wal").glob("*.json") \ + if (run_dir / "push_wal").is_dir() else (): + if _read_json(path).get("state") == "pushed": + pushed += 1 + for path in (run_dir / "ci_ops").glob("*.json") \ + if (run_dir / "ci_ops").is_dir() else (): + record = _read_json(path) + build_id = str(record.get("build_id") or "") + if build_id and record.get("state") in ("created", "cancelled"): + build_ids.add(build_id) + + durations: list[float] = [] + measured_ids: set[str] = set() + for round_ in ((substate.get("ci") or {}).get("rounds") or []): + build_id = str(round_.get("build_id") or "") + if build_id: + build_ids.add(build_id) + duration = round_.get("monitor_duration_sec") + if isinstance(duration, (int, float)) and duration >= 0: + durations.append(float(duration)) + if build_id: + measured_ids.add(build_id) + ci_partial = bool(build_ids - measured_ids) + ci_minutes = round(sum(durations) / 60.0, 4) if durations else ( + None if build_ids else 0.0) + return { + "pushes": pushed, + "ci_builds": len(build_ids), + "ci_minutes": ci_minutes, + "ci_minutes_partial": ci_partial, + } + + def _trace_events(run_dir: Path) -> list[dict]: """Read `run_dir/run_trace.jsonl` into a list of event dicts, skipping blank and malformed lines. Empty when the file is absent.""" @@ -346,24 +383,53 @@ def estimate_usd(tokens_in: int, tokens_out: int, ci_minutes: float, def _auto_components(kind: str, events: list[dict], progress: dict, - status: str) -> dict[str, float | None]: + status: str, *, substate: dict | None = None, + durable: dict | None = None) -> dict[str, float | None]: """The honestly auto-derivable slice of each task's components. Everything that needs juries, GT, CI re-runs, or maintainer feedback stays None.""" completed = progress.get("completed", {}) if isinstance(progress, dict) else {} components: dict[str, float | None] = { k: None for k in TASK_WEIGHTS.get(kind, {})} if kind in ("pr_rebase", "repo_rebase"): - if "push" in completed or ("rebase" in completed and status == "done"): + substate = substate or {} + durable = durable or {} + if ((status == "done" and (substate.get("phase") == "done" + or "finalize" in completed)) + or "push" in completed + or ("rebase" in completed and status == "done")): components["completed"] = 1.0 elif status in ("failed", "blocked"): components["completed"] = 0.0 # push_safe: guard violations surface as incidents; a completed push # with no push-related incident was a with-lease PR-head push by # construction (single choke point) - if "push" in completed: + if "push" in completed or durable.get("pushes", 0) > 0: components["push_safe"] = 1.0 if "verify" in completed and "gate" in completed: components["tests"] = 1.0 + modules = substate.get("modules") or {} + active_modules = [m or {} for m in modules.values() + if not (m or {}).get("skip")] + if active_modules: + components["conflict"] = 1.0 if all( + m.get("status") == "done" for m in active_modules) else 0.0 + tests = substate.get("tests") or {} + pipeline = tests.get("pipeline") or {} + precommit = (tests.get("precommit") or {}).get("result") + ci_result = (substate.get("ci") or {}).get("result") + test_failed = (int(pipeline.get("failed") or 0) > 0 + or bool(tests.get("infra_failures")) + or precommit == "failed" + or (ci_result and ci_result != "passed")) + if test_failed: + components["tests"] = 0.0 + elif ci_result == "passed" or ( + int(pipeline.get("passed") or 0) > 0 + and int(pipeline.get("skipped") or 0) == 0): + components["tests"] = 1.0 + if substate and components.get("completed") is not None: + components["purity"] = 0.0 if any( + ev.get("kind") == "out_of_scope_edit" for ev in events) else 1.0 elif kind == "pr_debug": # repro: fraction of debug groups whose output recorded tests_run # (foreach fan-out keys outputs by index; a single group is the @@ -400,38 +466,57 @@ def collect_run_metrics(run_dir: Path, settings: "Settings", events = _trace_events(run_dir) task = _read_json(run_dir / "task.json") progress = _read_json(run_dir / "progress.json") + substate = _read_json(run_dir / "substate.json") spec = task.get("spec") or task or {} kind = str(spec.get("kind") or "") ts = [float(ev["ts"]) for ev in events if isinstance(ev.get("ts"), (int, float))] minutes = (max(ts) - min(ts)) / 60.0 if len(ts) >= 2 else 0.0 - ci_minutes = sum(float(ev.get("minutes") or 0.0) for ev in events - if ev.get("kind") == "ci_build") + durable = _durable_rebase_signals(run_dir, substate) + event_ci_minutes = [float(ev.get("minutes") or 0.0) for ev in events + if ev.get("kind") == "ci_build" + and ev.get("minutes") is not None] + ci_minutes = (sum(event_ci_minutes) if event_ci_minutes + else durable["ci_minutes"]) + ci_minutes_source = ("trace" if event_ci_minutes else + "monitor_wall" if ci_minutes is not None + and durable["ci_builds"] else "none") + ci_time_partial = bool(durable["ci_minutes_partial"]) # ONE canonical cost source (W7): per-request llm spans when the trace # exists (each span priced by its own model + cache fields); the event # sum is the tagged fallback ONLY when no trace — never both span_cost = cost_from_spans(run_dir / "trace.jsonl", settings) - _, _, tool_calls = usage_from_events(events) + _, _, declared_tool_calls = usage_from_events(events) + explicit_tool_calls = sum(1 for ev in events + if ev.get("kind") == "tool_call") + tool_calls = max(declared_tool_calls, explicit_tool_calls) if span_cost is not None: tokens_in = span_cost["input_tokens"] tokens_out = span_cost["output_tokens"] - usd = span_cost["usd"] + ci_minutes * settings.ci_rate_usd_per_min + usd = (span_cost["usd"] + + (ci_minutes or 0.0) * settings.ci_rate_usd_per_min) cost_source = "spans" else: tokens_in, tokens_out, _ = usage_from_events(events) - usd = estimate_usd(tokens_in, tokens_out, ci_minutes, settings) + usd = estimate_usd(tokens_in, tokens_out, ci_minutes or 0.0, settings) cost_source = "events" incidents = derive_incidents(events) s = safety_multiplier(incidents) usd_ref = settings.cost_ref_usd.get(kind, settings.cost_ref_usd.get("default", 1.0)) min_ref = settings.cost_ref_min.get(kind, settings.cost_ref_min.get("default", 10.0)) - cost_partial = bool((span_cost or {}).get("cost_partial")) + llm_requests = sum(1 for ev in events + if ev.get("kind") == "rebase_llm_request") + usage_missing = span_cost is None and llm_requests > 0 \ + and tokens_in == 0 and tokens_out == 0 + cost_partial = bool((span_cost or {}).get("cost_partial")) \ + or ci_time_partial or usage_missing # a partial usd is a lower bound — a cost index over it would flatter the # run, so it is omitted rather than computed from a fabrication c = None if cost_partial else cost_index(usd, minutes, usd_ref, min_ref) - components = _auto_components(kind, events, progress, status) + components = _auto_components(kind, events, progress, status, + substate=substate, durable=durable) for key, value in (extra_components or {}).items(): components[key] = value abstained = _abstained(kind, events, status) @@ -455,7 +540,10 @@ def collect_run_metrics(run_dir: Path, settings: "Settings", "cost": { "usd": round(usd, 4), "minutes": round(minutes, 2), - "ci_minutes": round(ci_minutes, 2), + "ci_minutes": (round(ci_minutes, 2) + if ci_minutes is not None else None), + "ci_minutes_partial": ci_time_partial, + "ci_minutes_source": ci_minutes_source, "input_tokens": tokens_in, "output_tokens": tokens_out, "tool_calls": tool_calls, @@ -465,7 +553,8 @@ def collect_run_metrics(run_dir: Path, settings: "Settings", "cost_partial": cost_partial, "unpriced_calls": (span_cost or {}).get("unpriced_calls", 0), "source": cost_source, - "llm_calls": (span_cost or {}).get("llm_calls"), + "llm_calls": ((span_cost or {}).get("llm_calls") + if span_cost is not None else llm_requests or None), "by_role": (span_cost or {}).get("by_role"), "cache_read_tokens": (span_cost or {}).get("cache_read_tokens"), "cache_creation_tokens": (span_cost or {}).get("cache_creation_tokens"), @@ -479,7 +568,9 @@ def collect_run_metrics(run_dir: Path, settings: "Settings", "tus": None if (q is None or c is None) else round(tus(q, s, c), 4), "signals": { "escalations": sum(1 for ev in events if ev.get("kind") == "escalation"), - "pushes": sum(1 for ev in events if ev.get("kind") == "push_requested"), + "pushes": (durable["pushes"] or sum( + 1 for ev in events if ev.get("kind") == "push_requested")), + "ci_builds": durable["ci_builds"], "posted": sum(1 for ev in events if ev.get("kind") == "posted_artifact"), "steps_completed": len((progress or {}).get("completed", {})), }, diff --git a/src/infermatrix_copilot/notify.py b/src/infermatrix_copilot/notify.py index 6b9987ae4..94dac247c 100644 --- a/src/infermatrix_copilot/notify.py +++ b/src/infermatrix_copilot/notify.py @@ -67,6 +67,39 @@ def escalate(self, *, reason: str, phase: str, severity: str = "blocked", self._email(f"[infermatrix-copilot] {severity}: {phase}", body) return path + def resolve(self, resolution: str = "run completed successfully") -> bool: + """Archive a stale active escalation after a resumed run succeeds. + + ``ESCALATION.md`` is an active-attention marker. Keeping it beside a + successful terminal status is misleading, but deleting it would lose + useful history, so its full contents are appended to the history file + before the marker is removed. + """ + active = self.run_dir / "ESCALATION.md" + if not active.is_file(): + return False + try: + body = active.read_text(encoding="utf-8") + history = self.run_dir / "ESCALATION_HISTORY.md" + previous = (history.read_text(encoding="utf-8") + if history.is_file() else "") + # Idempotent across a crash after the history append but before + # unlinking the active marker. + if body.rstrip() not in previous: + with open(history, "a", encoding="utf-8") as stream: + if previous: + stream.write("\n---\n\n") + stream.write(body.rstrip()) + stream.write("\n\n## Resolution\n") + stream.write( + f"- **when**: {time.strftime('%Y-%m-%d %H:%M:%S')}\n") + stream.write(f"- {resolution}\n") + active.unlink() + except OSError: + return False + self.trace.record("escalation_resolved", resolution=resolution) + return True + def _render(self, esc: Escalation) -> str: """Render an Escalation into the ESCALATION.md Markdown body: run/phase/ timestamp header, the reason, the state summary as a JSON block, and an diff --git a/src/infermatrix_copilot/rebase_engine/ci_loop.py b/src/infermatrix_copilot/rebase_engine/ci_loop.py index 940a71aed..98d67869a 100644 --- a/src/infermatrix_copilot/rebase_engine/ci_loop.py +++ b/src/infermatrix_copilot/rebase_engine/ci_loop.py @@ -781,6 +781,7 @@ class CIBuildRound: ignored: int = 0 ignored_baseline: int = 0 jobs: list[dict] = field(default_factory=list) + monitor_duration_sec: float | None = None @dataclass @@ -1143,6 +1144,7 @@ def _trace(event: str, **fields) -> None: rounds.append(rec) _trace("ci_build", round=rnd, purpose=rec.purpose, build=rec.build_id, adopted=rec.adopted) + monitor_started = time.monotonic() outcome = await asyncio.to_thread( monitor_build, client, rec.build_id, spec=spec, poll_sec=poll_sec, timeout_sec=timeout_sec, @@ -1150,6 +1152,8 @@ def _trace(event: str, **fields) -> None: # run cannot prove it created (round-4 review) retry_max=0 if rec.adopted else job_retry_max, log_dir=log_dir, sleep=sleep, now=now) + rec.monitor_duration_sec = round( + max(0.0, time.monotonic() - monitor_started), 3) if outcome.no_run: rec.build_state = outcome.build_state _trace("ci_build_refused", build=rec.build_id, @@ -1170,11 +1174,14 @@ def _trace(event: str, **fields) -> None: adopted=True) rounds.append(rec) _trace("ci_sibling_adopted", build=rec.build_id) + monitor_started = time.monotonic() outcome = await asyncio.to_thread( monitor_build, client, rec.build_id, spec=spec, poll_sec=poll_sec, timeout_sec=timeout_sec, retry_max=0, log_dir=log_dir, sleep=sleep, now=now) + rec.monitor_duration_sec = round( + max(0.0, time.monotonic() - monitor_started), 3) _record_outcome(rec, outcome) if outcome.clean_pass: return CIRunResult("passed", rounds=rounds, diff --git a/test/test_ci_wiring.py b/test/test_ci_wiring.py index d2cf80646..93c5701e0 100644 --- a/test/test_ci_wiring.py +++ b/test/test_ci_wiring.py @@ -769,6 +769,7 @@ def cancel_build(self, build_id): assert fresh[0].state == "created" # the monitored build is the FRESH one at the new commit assert result.rounds[0].build_id == fresh[0].build_id + assert result.rounds[0].monitor_duration_sec is not None assert result.rounds[0].jobs == [{ "name": "Green", "job_id": "Green", "state": "passed", "exit_status": 0, "classification": "passed", "log_file": "", diff --git a/test/test_curator.py b/test/test_curator.py index db0dd25a8..271955676 100644 --- a/test/test_curator.py +++ b/test/test_curator.py @@ -434,9 +434,16 @@ def test_phase5_report_and_compare_steps(settings, trace, tmp_path): run_dir.mkdir() sub = Substate(run_dir, "run-p") sub.update({"modules": {"core": {"status": "done"}, + "docs": {"skip": True}, "worker": {"status": "failed"}}, "tests": {"pipeline": {"passed": 3, "failed": 1, - "failed_tests": ["t_x"]}}}) + "failed_tests": ["t_x"], + "skipped": 2}}, + "ci": {"result": "passed", "reason": "baseline debt", + "rounds": [{"build_id": "42", + "build_state": "failed", "passed": 8, + "failed": [], "ignored": 1, + "ignored_baseline": 2}]}}) def ctx(params=None): return StepContext( @@ -446,12 +453,26 @@ def ctx(params=None): "repo": "widget-repo", "params": {"rebase_mode": "full", **(params or {})}}, - "repo_path": str(repo), "run_id": "run-p"}) + "repo_path": str(repo), "run_id": "run-p", + "manifest_jobs": 6}) r = asyncio.run(registry.get("rebase.v3_phase5_report").handler(ctx())) assert r.ok summary = (run_dir / "FINAL_SUMMARY.md").read_text(encoding="utf-8") assert "worker: failed" in summary and "failed: t_x" in summary + assert "docs: skipped (not assigned)" in summary + assert "manifest jobs: 6" in summary and "not run/skipped: 2" in summary + assert "effective result: passed" in summary + assert "raw provider state: failed (build 42" in summary + + # The generic report is the canonical user-facing artifact; for a rebase + # it embeds the phase-5 result rather than producing an empty shell. + r = asyncio.run(registry.get("report.final_summary").handler(ctx())) + assert r.ok + report = (run_dir / "RUN_REPORT.md").read_text(encoding="utf-8") + assert "## Rebase result" in report + assert "docs: skipped (not assigned)" in report + assert "raw provider state: failed" in report # compare with no baseline and no declared knowledge: artifact + # explicit no-baseline note, no drift machinery diff --git a/test/test_engine.py b/test/test_engine.py index e7adc1dda..55abbc2a6 100644 --- a/test/test_engine.py +++ b/test/test_engine.py @@ -128,6 +128,21 @@ def test_blocked_escalates(env, tmp_path): assert (tmp_path / "run" / "ESCALATION.md").exists() +def test_resolved_escalation_moves_to_history(env, tmp_path): + registry, executor, notifier = env + registry.register(make_step( + "s.blocked", lambda ctx: StepResult( + False, FailureKind.BLOCKED, "temporary operator action"))) + asyncio.run(executor.run(playbook([PlaybookStep("a", "s.blocked")]), {})) + + assert notifier.resolve("resumed workflow completed") is True + assert not (tmp_path / "run" / "ESCALATION.md").exists() + history = (tmp_path / "run" / "ESCALATION_HISTORY.md").read_text() + assert "temporary operator action" in history + assert "resumed workflow completed" in history + assert notifier.resolve() is False + + def test_escalation_summary_and_artifacts_forwarded(env): registry, executor, notifier = env registry.register(make_step( diff --git a/test/test_metrics.py b/test/test_metrics.py index dc018a0d3..88022d493 100644 --- a/test/test_metrics.py +++ b/test/test_metrics.py @@ -208,6 +208,59 @@ def test_collect_rebase_auto_components_and_partial_flag(settings, tmp_path): assert m["catq"] is not None and 0 < m["catq"] <= 1.0 +def test_collect_repo_rebase_uses_durable_operations_and_substate( + settings, tmp_path): + run_dir = _make_run(tmp_path, "repo_rebase", events=[ + {"kind": "task"}, + {"kind": "rebase_llm_request"}, + {"kind": "tool_call", "tool": "read_file"}, + {"kind": "tool_call", "tool": "run_shell"}, + ], completed={"finalize": {}}) + (run_dir / "push_wal").mkdir() + (run_dir / "push_wal" / "p0.json").write_text(json.dumps({ + "state": "pushed"})) + (run_dir / "ci_ops").mkdir() + (run_dir / "ci_ops" / "c0.json").write_text(json.dumps({ + "state": "created", "build_id": "42"})) + (run_dir / "substate.json").write_text(json.dumps({ + "phase": "done", + "modules": {"active": {"status": "done", "skip": False}, + "idle": {"skip": True}}, + "tests": {"pipeline": {"passed": 0, "failed": 0, "skipped": 3}, + "precommit": {"result": "failed_preexisting"}}, + "ci": {"result": "passed", "rounds": [{ + "build_id": "42", "monitor_duration_sec": 150.0}]}, + })) + + metrics = collect_run_metrics(run_dir, settings, "done") + assert metrics["signals"]["pushes"] == 1 + assert metrics["signals"]["ci_builds"] == 1 + assert metrics["cost"]["ci_minutes"] == 2.5 + assert metrics["cost"]["ci_minutes_source"] == "monitor_wall" + assert metrics["cost"]["tool_calls"] == 2 + assert metrics["cost"]["llm_calls"] == 1 + assert metrics["quality"]["components"] == { + "completed": 1.0, "conflict": 1.0, "tests": 1.0, + "purity": 1.0, "push_safe": 1.0, + } + # Tokens were not exposed for the durable LLM request, so USD/CATQ stay + # explicitly partial instead of pretending the run cost nothing. + assert metrics["cost"]["cost_partial"] is True + assert metrics["cost"]["cost_index"] is None + + +def test_collect_repo_rebase_marks_unmeasured_ci_time_unknown(settings, + tmp_path): + run_dir = _make_run(tmp_path, "repo_rebase", events=[{"kind": "task"}]) + (run_dir / "ci_ops").mkdir() + (run_dir / "ci_ops" / "c0.json").write_text(json.dumps({ + "state": "cancelled", "build_id": "old"})) + metrics = collect_run_metrics(run_dir, settings, "blocked") + assert metrics["cost"]["ci_minutes"] is None + assert metrics["cost"]["ci_minutes_partial"] is True + assert metrics["cost"]["cost_partial"] is True + + def test_collect_rebase_safe_abstain_scores_fixed(settings, tmp_path): run_dir = _make_run(tmp_path, "pr_rebase", events=[ {"kind": "task"},