"""T10 轨迹回放与会议结果复盘。 依据:docs/v2/03-detailed-design.md §9.1/§9.4、14-experience-loop.md §7/§8; BR-13/31/32、E-05/23/26、AC-22/32/33。 三条不可让渡的规则: 1. **只读回放**:轨迹从既有 run 的步骤投影出来,本模块不写 run,也不改历史结果。 2. **失败保留**:失败的步骤照样出现并带错误码,不因为"看起来不好"被过滤。 3. **结果追加**:会议结果只能新增;更正生成新记录并指向 `supersedes_id`, 找不到材料版本的一律记 unmatched 并单列,不当成否决、也不进通过率分母。 """ from __future__ import annotations from contracts.prescreen import EventOutcome from contracts.trajectory import ReviewOutcome, TrajectoryStep from services import assistant_store as store KIND = "review_outcome" DEFAULT_PAGE = 50 # 既有步骤状态 → 轨迹状态。partial 不是完成,保留原样说明,避免看起来"全过"。 _STATUS_MAP = {"running": "running", "completed": "completed", "partial": "completed", "failed": "failed", "skipped": "skipped", "cancelled": "cancelled"} _PARTIAL_NOTE = "部分完成(原始状态 partial);未完成的部分不计入已核查" _FAILURE_CODE_FALLBACK = "RUN_STEP_FAILED" RAW_ALIASES = { "通过": EventOutcome.approved, "同意": EventOutcome.approved, "否决": EventOutcome.rejected, "未通过": EventOutcome.rejected, "附条件": EventOutcome.conditional, "附条件通过": EventOutcome.conditional, "暂缓": EventOutcome.deferred, "撤回": EventOutcome.withdrawn, "未送审": EventOutcome.not_submitted, } def _step(run_body: dict, raw: dict, index: int) -> TrajectoryStep: status = _STATUS_MAP.get(str(raw.get("status")), "skipped") errors = raw.get("errors") or () notes = [] if raw.get("status") == "partial": notes.append(_PARTIAL_NOTE) if raw.get("missing_inputs"): notes.append("缺证:" + "、".join(str(item) for item in raw["missing_inputs"])) skill = raw.get("skill") or {} component = {"id": skill["id"], "version": str(skill.get("version", ""))} if skill.get("id") else None return TrajectoryStep( id=str(raw.get("step_id")), run_id=str(raw.get("run_id") or run_body.get("id")), parent_step_id=None, sequence=index, kind="skill" if raw.get("skill") else "tool" if raw.get("tool") else "plan", component_ref=component, input_refs=[str(raw.get("input_snapshot_id"))] if raw.get("input_snapshot_id") else [], output_refs=sorted(str(key) for key in (raw.get("outputs") or {})), evidence_ids=sorted({str(item) for finding in raw.get("findings") or () for item in (finding.get("evidence_ids") or ())}), status=status, started_at=raw.get("started_at"), finished_at=raw.get("finished_at"), reason_summary=";".join(notes) or None, error_code=(str(errors[0].get("code")) if errors and errors[0].get("code") else _FAILURE_CODE_FALLBACK) if status == "failed" else None, latency_ms=raw.get("latency_ms"), cost=None, cost_currency=None) # 成本缺省一律留空:既有运行没有成本口径时不得填 0 或估计值(06 §5)。 def project_steps(run_body: dict) -> list[dict]: """把既有 run 的步骤投影成轨迹步骤,顺序即 sequence,不改写原 run。""" return [_step(run_body, raw, index).model_dump(mode="json") for index, raw in enumerate(run_body.get("steps") or ())] def replay(db, project_id: str, run_id: str, cursor: str | None = None, limit: int = DEFAULT_PAGE) -> dict: """只读回放:按 sequence 稳定分页。不存在或跨项目 → KeyError(接口层转 404)。""" run = store.get(db, "run", run_id) if run.get("project_id") != project_id: raise KeyError("运行不存在或不属于该项目") steps = project_steps(run) start = max(0, int(cursor or 0)) window = steps[start:start + max(1, min(limit, 200))] end = start + len(window) return { "run_id": run_id, "project_id": project_id, "status": run.get("status"), "snapshot_id": run.get("input_snapshot_id"), "steps": window, "next_cursor": str(end) if end < len(steps) else None, "read_only": True, "note": "轨迹只记录可观察动作与版本;不包含模型内部推理过程", } def _outcomes(db, project_id: str) -> list[dict]: return list(store.listing(db, KIND, project_id)) def _chain_has_cycle(rows: list[dict], start: str) -> bool: """防御性检查:追加写入本身造不出环,但迁移/导入的旧行可能有(03 §9.1)。""" seen, current = set(), start while current: if current in seen: return True seen.add(current) current = next((item.get("supersedes_id") for item in rows if item["id"] == current), None) return False def record(db, project_id: str, body: dict, actor: str, known_snapshots=()) -> dict: """追加一条会议结果。更正只能新增,且必须指向同项目的旧记录。""" rows = _outcomes(db, project_id) raw_code = str(body.get("raw_code") or "").strip() if not raw_code: raise ValueError("会议结果必须给出原始结果码,未提供时不得推断") normalized = body.get("normalized") or RAW_ALIASES.get(raw_code, EventOutcome.unknown) if isinstance(normalized, str): normalized = EventOutcome(normalized) snapshot = body.get("reviewed_snapshot_id") matched = bool(snapshot) and (not known_snapshots or snapshot in set(known_snapshots)) supersedes = body.get("supersedes_id") if supersedes: previous = next((item for item in rows if item["id"] == supersedes), None) if previous is None: raise ValueError("更正要指向本项目既有的会议结果记录") if _chain_has_cycle(rows, supersedes): raise ValueError("更正链出现循环,已拒绝") mapping_ref = body.get("mapping_ref") label = None if matched and mapping_ref and normalized in (EventOutcome.approved, EventOutcome.rejected): label = normalized == EventOutcome.approved revision = 1 + max([item.get("revision", 0) for item in rows if item.get("meeting_id") == body.get("meeting_id")] or [0]) outcome = ReviewOutcome( id=store.uid("outcome"), project_id=project_id, meeting_id=str(body.get("meeting_id") or ""), revision=revision, occurred_at=body.get("occurred_at") or store.now(), recorded_at=store.now(), reviewed_snapshot_id=snapshot if matched else None, raw_code=raw_code, normalized=normalized, mapping_ref=mapping_ref, label=label, conditions=list(body.get("conditions") or ()), evidence_ids=list(body.get("evidence_ids") or ()), status="matched" if matched else "unmatched", supersedes_id=supersedes, recorded_by=actor) saved = outcome.model_dump(mode="json") store.put(db, KIND, saved) store.event(db, project_id, "review_outcome.recorded", actor, {"outcome_id": saved["id"], "status": saved["status"], "raw_code": saved["raw_code"]}, None) return saved def listing(db, project_id: str, status: str | None = None) -> dict: """按匹配状态分组返回。unmatched 单列,不混进通过率分母。""" rows = sorted(_outcomes(db, project_id), key=lambda item: item.get("recorded_at") or "") observed = [item for item in rows if item["status"] == "matched"] unmatched = [item for item in rows if item["status"] == "unmatched"] if status == "matched": return {"items": observed} if status == "unmatched": return {"items": unmatched} return {"items": rows, "matched": observed, "unmatched": unmatched, "counts_toward_denominator": denominator(rows)} def denominator(rows) -> dict: """已观察分母只数"匹配到材料版本且有通过/否决口径"的记录(AC-22/37)。""" labelled = [item for item in rows if item.get("status") == "matched" and item.get("normalized") in ("approved", "rejected")] excluded = [(item["id"], item.get("raw_code"), item.get("normalized")) for item in rows if item not in labelled] return {"observed": len(labelled), "excluded": excluded, "note": "未送审、撤回、附条件与未知结果不计入已观察通过率;不把未知当否决"}