deerflow-code/offline-backend-20260512/backend/app/maf_roundtable/quality_contracts.py

"""Stage 4 (R3-Q): seat task contracts, deterministic quality checks, manager
review schema, and server-owned decision finalization.

Pure policy module: no I/O and no MAF SDK imports, so the whole acceptance
policy is unit-testable with a weak/fake model before any workflow runs.

The manager model only ever *proposes* review decisions. This module is the
final validator and never allows:

- accepting a delivery that failed the deterministic quality checks (the
  manager's self-assessment is NOT acceptance evidence);
- reworking a seat beyond its frozen ``max_rework_attempts``;
- more reassignments than the server-configured cap;
- an exhausted task being disguised as accepted — it becomes a
  ``controlled_gap`` carrying its risk, or an explicit ``ask_user`` pause.

Task status state machine (persisted per seat task)::

    assigned → delivering → submitted → manager_review →
        accepted
        | rework        (→ delivering, with explicit revision requirements)
        | reassigned    (→ delivering, backup seat takes over)
        | exhausted     (rework/reassign caps reached)
            → controlled_gap (risk recorded, carried into the final report)
"""

from __future__ import annotations

import hashlib
import json
import re
from typing import Any, Literal

from pydantic import BaseModel, ConfigDict, Field

# ── task status state machine ───────────────────────────────────────────────

TASK_ASSIGNED = "assigned"
TASK_DELIVERING = "delivering"
TASK_SUBMITTED = "submitted"
TASK_MANAGER_REVIEW = "manager_review"
TASK_ACCEPTED = "accepted"
TASK_REWORK = "rework"
TASK_REASSIGNED = "reassigned"
TASK_EXHAUSTED = "exhausted"
TASK_CONTROLLED_GAP = "controlled_gap"

TASK_TERMINAL_STATUSES = frozenset({TASK_ACCEPTED, TASK_CONTROLLED_GAP})
TASK_ALL_STATUSES = frozenset({
    TASK_ASSIGNED,
    TASK_DELIVERING,
    TASK_SUBMITTED,
    TASK_MANAGER_REVIEW,
    TASK_ACCEPTED,
    TASK_REWORK,
    TASK_REASSIGNED,
    TASK_EXHAUSTED,
    TASK_CONTROLLED_GAP,
})

MafReviewDecision = Literal[
    "accept", "return_for_rework", "reassign", "ask_user", "controlled_gap"
]
MafFallbackPolicy = Literal["reassign_then_gap", "controlled_gap", "ask_user"]

# Final (server-owned) outcomes of the decision finalizer.
OUTCOME_ACCEPTED = "accepted"
OUTCOME_REWORK = "rework"
OUTCOME_REASSIGNED = "reassigned"
OUTCOME_CONTROLLED_GAP = "controlled_gap"
OUTCOME_ASK_USER = "ask_user"


class InvalidTaskContractError(ValueError):
    """Raised when a task contract cannot be frozen from a dispatch plan."""


# ── task contract ───────────────────────────────────────────────────────────


class MafTaskContract(BaseModel):
    """Frozen acceptance contract for one seat task.

    Created server-side from the validated dispatch assignment before the seat
    starts working. The manager model can never widen these terms afterwards.
    """

    model_config = ConfigDict(extra="forbid")

    objective: str = Field(min_length=1, max_length=2000)
    expected_deliverable: str = Field(min_length=1, max_length=2000)
    acceptance_criteria: list[str] = Field(default_factory=list, max_length=32)
    required: bool = True
    max_rework_attempts: int = Field(default=2, ge=0, le=8)
    fallback_policy: MafFallbackPolicy = "reassign_then_gap"


DEFAULT_EXPECTED_DELIVERABLE = (
    "结构化 JSON 交付：结论(summary)、依据/证据(findings/evidence_refs)、"
    "未解决项(open_questions)、自检(self_check)"
)


def build_task_contract(
    *,
    task_goal: str,
    delivery_requirements: list[str] | None = None,
    required: bool = True,
    max_rework_attempts: int = 2,
    fallback_policy: MafFallbackPolicy = "reassign_then_gap",
) -> MafTaskContract:
    """Freeze a task contract from one validated dispatch assignment."""
    objective = str(task_goal or "").strip()
    if not objective:
        raise InvalidTaskContractError("task contract objective must be non-empty")
    criteria = [
        str(item).strip()
        for item in (delivery_requirements or [])
        if str(item or "").strip()
    ]
    return MafTaskContract(
        objective=objective,
        expected_deliverable=DEFAULT_EXPECTED_DELIVERABLE,
        acceptance_criteria=criteria,
        required=required,
        max_rework_attempts=max_rework_attempts,
        fallback_policy=fallback_policy,
    )


# ── seat delivery ───────────────────────────────────────────────────────────


class MafSeatDelivery(BaseModel):
    """Structured delivery extracted from one seat turn output."""

    model_config = ConfigDict(extra="forbid")

    conclusion: str = ""
    evidence: list[str] = Field(default_factory=list)
    # P1: only ``evidence_refs`` can satisfy the evidence requirement. Seat
    # ``findings`` are candidate claims — they are kept in ``evidence`` for
    # review display but never count as verified evidence.
    evidence_refs: list[str] = Field(default_factory=list)
    unresolved_items: list[str] = Field(default_factory=list)
    self_check: str = ""


def _flat_text(item: Any, *keys: str) -> str:
    if isinstance(item, dict):
        for key in keys:
            value = item.get(key)
            if isinstance(value, str) and value.strip():
                return value.strip()
        return ""
    return str(item or "").strip()


def delivery_from_seat_output(
    *, structured_output: dict[str, Any] | None, raw_content: str
) -> MafSeatDelivery:
    """Map a validated SeatTurnOutput dict onto the delivery contract.

    ``structured_output`` is ``None`` when the seat output failed the schema —
    the resulting delivery is (almost) empty so the deterministic checks flag
    it instead of trusting the raw text.
    """
    out = structured_output if isinstance(structured_output, dict) else {}
    conclusion = str(out.get("summary") or "").strip()

    evidence: list[str] = []
    for item in out.get("findings") or []:
        text = _flat_text(item, "text")
        if text:
            evidence.append(text)
    evidence_refs: list[str] = []
    for ref in out.get("evidence_refs") or []:
        text = _flat_text(ref, "ref", "text")
        if text:
            evidence_refs.append(text)
            evidence.append(text)

    unresolved = [
        str(q).strip() for q in (out.get("open_questions") or []) if str(q or "").strip()
    ]
    self_check = str(out.get("self_check") or "").strip()
    return MafSeatDelivery(
        conclusion=conclusion,
        evidence=evidence,
        evidence_refs=evidence_refs,
        unresolved_items=unresolved,
        self_check=self_check,
    )


# ── deterministic quality checks ────────────────────────────────────────────

FAIL_EMPTY_CONTENT = "empty_content"
FAIL_SCHEMA_INVALID = "schema_invalid"
FAIL_MISSING_CONCLUSION = "missing_conclusion"
FAIL_BELOW_MIN_LENGTH = "below_min_length"
FAIL_MISSING_EVIDENCE = "missing_evidence"
FAIL_DUPLICATE_CONTENT = "duplicate_content"
FAIL_UNVERIFIED_EVIDENCE_REF = "unverified_evidence_ref"

_REPAIR_HINTS: dict[str, str] = {
    FAIL_EMPTY_CONTENT: "交付内容为空，请给出完整的分析与结论",
    FAIL_SCHEMA_INVALID: "输出不是合法的结构化 JSON，请按 SeatTurnOutput 字段重新输出",
    FAIL_MISSING_CONCLUSION: "缺少明确结论（summary），请补充",
    FAIL_BELOW_MIN_LENGTH: "结论过短，请展开具体分析、依据与建议",
    FAIL_MISSING_EVIDENCE: (
        "缺少可信证据（evidence_refs）。findings 只是候选主张，不能充当证据；"
        "请引用事实账本/成功工具结果中真实存在的证据引用"
    ),
    FAIL_DUPLICATE_CONTENT: "内容与上一次交付重复，请针对退回意见提供新的分析",
    FAIL_UNVERIFIED_EVIDENCE_REF: (
        "引用了事实账本中不存在的证据（自造引用不可接受）；"
        "请只引用已记录的可信证据或成功工具结果"
    ),
}


def conclusion_digest(conclusion: str) -> str:
    """SHA-256 digest of the normalized conclusion.

    Duplicate detection persists these digests (never the conclusion bodies)
    in the task history, so the check survives restarts without storing the
    duplicated text again.
    """
    normalized = " ".join(str(conclusion or "").split())
    return hashlib.sha256(normalized.encode("utf-8")).hexdigest()


def check_delivery_deterministic(
    *,
    delivery: MafSeatDelivery,
    raw_content: str,
    schema_status: str | None,
    min_length: int = 40,
    min_evidence: int = 1,
    prior_conclusion_digests: set[str] | list[str] | None = None,
    verified_evidence_refs: set[str] | None = None,
) -> dict[str, Any]:
    """Deterministic acceptance gate run BEFORE the manager review.

    A delivery that fails any check can never be accepted, regardless of what
    the manager model claims. Returns ``{"passed": bool, "failure_codes": [...]}``.

    Evidence rules (P1):
    - seat ``findings`` are candidate claims only — they can NEVER satisfy the
      evidence requirement; only ``evidence_refs`` count;
    - when ``verified_evidence_refs`` is supplied (the trusted refs the fact
      ledger actually backs: recorded evidence + successful tool results),
      every cited ref must resolve — an invented ref fails with
      ``unverified_evidence_ref``, and only verified refs count toward
      ``min_evidence``;
    - duplicate detection compares SHA-256 digests of normalized conclusions,
      so it survives restarts via the persisted task history.
    """
    codes: list[str] = []
    text = str(raw_content or "").strip()
    if not text:
        codes.append(FAIL_EMPTY_CONTENT)
    if schema_status == "schema_invalid":
        codes.append(FAIL_SCHEMA_INVALID)
    if not delivery.conclusion:
        codes.append(FAIL_MISSING_CONCLUSION)
    elif len(delivery.conclusion) < max(0, int(min_length)):
        codes.append(FAIL_BELOW_MIN_LENGTH)

    cited_refs = [str(r) for r in (delivery.evidence_refs or []) if str(r or "").strip()]
    if verified_evidence_refs is not None:
        trusted = {str(r) for r in verified_evidence_refs}
        verified = [r for r in cited_refs if r in trusted]
        if any(r not in trusted for r in cited_refs):
            codes.append(FAIL_UNVERIFIED_EVIDENCE_REF)
    else:
        verified = cited_refs
    if len(verified) < max(0, int(min_evidence)):
        codes.append(FAIL_MISSING_EVIDENCE)

    if (
        delivery.conclusion
        and prior_conclusion_digests
        and conclusion_digest(delivery.conclusion) in set(prior_conclusion_digests)
    ):
        codes.append(FAIL_DUPLICATE_CONTENT)
    return {"passed": not codes, "failure_codes": codes}


def repair_requirements_for_codes(failure_codes: list[str]) -> list[str]:
    """Deterministic, targeted revision requirements for failed checks."""
    hints = [_REPAIR_HINTS[code] for code in failure_codes if code in _REPAIR_HINTS]
    return hints or ["请按照任务合同补充缺失内容后重新交付"]


# ── manager review schema ───────────────────────────────────────────────────


class MafManagerReviewItem(BaseModel):
    """One per-task review decision proposed by the manager model."""

    model_config = ConfigDict(extra="forbid")

    agent_id: str = Field(min_length=1, max_length=64)
    decision: MafReviewDecision
    reason: str = Field(min_length=1, max_length=1000)
    revision_requirements: list[str] = Field(default_factory=list, max_length=16)


class MafManagerReviewOutput(BaseModel):
    """Strict model-facing manager review schema (one entry per reviewed seat)."""

    model_config = ConfigDict(extra="forbid")

    decisions: list[MafManagerReviewItem] = Field(default_factory=list, max_length=64)


_JSON_FENCE_RE = re.compile(r"```(?:json)?\s*([\s\S]*?)```", re.IGNORECASE)


def _extract_json_object(text: str) -> str | None:
    raw = (text or "").strip()
    if not raw:
        return None
    m = _JSON_FENCE_RE.search(raw)
    if m:
        raw = m.group(1).strip()
    if raw.startswith("{") and raw.endswith("}"):
        return raw
    start = raw.find("{")
    end = raw.rfind("}")
    if start >= 0 and end > start:
        return raw[start : end + 1]
    return None


def parse_manager_review(text: str) -> tuple[MafManagerReviewOutput | None, list[str]]:
    """Parse the manager's review JSON. Returns ``(model, readable_errors)``."""
    blob = _extract_json_object(text)
    if blob is None:
        return None, ["review output is not a JSON object; emit MafManagerReviewOutput JSON"]
    try:
        data = json.loads(blob)
    except json.JSONDecodeError as exc:
        return None, [f"invalid JSON: {exc.msg}"]
    if not isinstance(data, dict):
        return None, ["JSON root must be an object"]
    try:
        return MafManagerReviewOutput.model_validate(data), []
    except Exception as exc:  # pydantic ValidationError
        errors = []
        detail = getattr(exc, "errors", None)
        if callable(detail):
            for err in detail():
                loc = ".".join(str(p) for p in err.get("loc", ()))
                errors.append(f"{loc}: {err.get('msg', 'invalid')}")
        return None, errors or ["manager review schema validation failed"]


def build_manager_review_prompt(
    *,
    task_text: str,
    review_items: list[dict[str, Any]],
) -> str:
    """Build the strict manager review prompt.

    Each review item: ``{agent_id, attempt, contract, delivery, check}``. The
    manager may only propose decisions; the server finalizes them.
    """
    contract_shape = {
        "decisions": [
            {
                "agent_id": "被审查席位 id",
                "decision": "accept|return_for_rework|reassign|ask_user|controlled_gap",
                "reason": "必填：针对性理由",
                "revision_requirements": ["打回时必填：缺什么、怎样补、哪些原结论不可用"],
            }
        ]
    }
    lines = [
        "你是 MAF 总控，负责验收下列席位交付。只输出一个 JSON 对象（不要 Markdown），"
        "对每个席位给出一条 decision。验收依据是任务合同与确定性检查结果；"
        "确定性检查未通过的交付不得 accept。打回时必须写明缺什么、怎样补。"
        "不要输出任何席位/工具/线程 id 之外的字段。",
        f"原始任务：\n{task_text[:2000]}",
        "待审查交付：",
        json.dumps(review_items, ensure_ascii=False)[:12000],
        f"必须返回的 JSON 结构：{json.dumps(contract_shape, ensure_ascii=False)}",
    ]
    return "\n\n".join(lines)


# ── server-owned decision finalization ──────────────────────────────────────


def finalize_review_decision(
    *,
    contract: MafTaskContract,
    check_result: dict[str, Any],
    review_item: MafManagerReviewItem | None,
    rework_count: int,
    reassign_count: int,
    has_backup_seat: bool,
    max_reassignments: int = 1,
) -> dict[str, Any]:
    """Finalize one task's review decision server-side.

    Returns ``{"decision", "reason", "revision_requirements", "source"}`` where
    ``decision`` is one of the final outcomes: ``accepted`` / ``rework`` /
    ``reassigned`` / ``controlled_gap`` / ``ask_user``.

    Rules enforced here (never by the model alone):

    1. A deterministic check failure can NEVER end in ``accepted``.
    2. Rework is only granted while ``rework_count < contract.max_rework_attempts``.
    3. Reassignment is only granted while a backup seat exists and
       ``reassign_count < max_reassignments``.
    4. Once every cap is exhausted the task ends as ``controlled_gap`` (or an
       explicit ``ask_user`` pause when the contract's fallback policy says
       so) — never as a disguised success.
    5. A missing/invalid manager review (timeout, exception, empty output,
       invalid JSON, or a missing per-seat decision) is NEVER an acceptance,
       even when the delivery passed every deterministic check: while the
       rework cap allows it, the server forces a system rework; once the caps
       are exhausted the contract's fallback policy decides (reassign /
       ``controlled_gap`` / ``ask_user``).
    """
    codes = list(check_result.get("failure_codes") or [])
    deterministic_passed = bool(check_result.get("passed"))
    can_rework = rework_count < contract.max_rework_attempts
    can_reassign = has_backup_seat and reassign_count < max_reassignments

    def _outcome(decision: str, reason: str, source: str, requirements: list[str] | None = None) -> dict[str, Any]:
        return {
            "decision": decision,
            "reason": str(reason or "")[:1000],
            "revision_requirements": [str(r) for r in (requirements or [])][:16],
            "source": source,
        }

    def _exhaustion_outcome(reason: str, source: str) -> dict[str, Any]:
        if contract.fallback_policy == "ask_user":
            return _outcome(OUTCOME_ASK_USER, reason, source)
        if contract.fallback_policy == "reassign_then_gap" and can_reassign:
            return _outcome(OUTCOME_REASSIGNED, reason, source)
        return _outcome(OUTCOME_CONTROLLED_GAP, reason, source)

    # 1) Deterministic failure — never accepted.
    if not deterministic_passed:
        base_reason = f"deterministic_quality_check_failed: {', '.join(codes)}"
        if can_rework:
            requirements = (
                list(review_item.revision_requirements)
                if review_item is not None
                and review_item.decision == "return_for_rework"
                and review_item.revision_requirements
                else repair_requirements_for_codes(codes)
            )
            return _outcome(OUTCOME_REWORK, base_reason, "deterministic", requirements)
        return _exhaustion_outcome(base_reason, "deterministic")

    # 2) Deterministic checks passed — but acceptance still requires a valid
    # manager review. A missing/invalid review (timeout, exception, empty
    # output, invalid JSON, or a missing per-seat decision) is NEVER an
    # acceptance: passing length/format checks alone does not make a delivery
    # verified. The server forces a system rework while the cap allows it, and
    # falls back to the contract's exhaustion policy afterwards.
    if review_item is None:
        reason = "manager_review_missing_or_invalid"
        if can_rework:
            return _outcome(
                OUTCOME_REWORK,
                reason,
                "system",
                ["总控未获得有效验收结论；请重新提交完整、可核验的交付。"],
            )
        return _exhaustion_outcome(reason, "system")

    decision = review_item.decision
    if decision == "accept":
        return _outcome(OUTCOME_ACCEPTED, review_item.reason, "manager")
    if decision == "return_for_rework":
        if can_rework:
            requirements = list(review_item.revision_requirements) or repair_requirements_for_codes(codes)
            return _outcome(OUTCOME_REWORK, review_item.reason, "manager", requirements)
        return _exhaustion_outcome(
            f"rework_cap_reached: {review_item.reason}", "manager"
        )
    if decision == "reassign":
        if can_reassign:
            return _outcome(OUTCOME_REASSIGNED, review_item.reason, "manager")
        return _exhaustion_outcome(
            f"reassign_unavailable: {review_item.reason}", "manager"
        )
    if decision == "ask_user":
        return _outcome(OUTCOME_ASK_USER, review_item.reason, "manager")
    # decision == "controlled_gap"
    return _outcome(OUTCOME_CONTROLLED_GAP, review_item.reason, "manager")


def make_controlled_gap(
    *,
    task_key: str,
    agent_id: str,
    objective: str,
    reason: str,
    failure_codes: list[str] | None = None,
    attempts: int = 0,
) -> dict[str, Any]:
    """Build the persisted, traceable controlled-gap record.

    A controlled gap is NEVER a verified fact: the final report must carry it
    in its risk / pending-information section.
    """
    return {
        "task_key": task_key,
        "agent_id": agent_id,
        "objective": str(objective or "")[:2000],
        "reason": str(reason or "")[:1000],
        "failure_codes": list(failure_codes or []),
        "attempts": int(attempts),
        "risk": "该问题未获得合格席位交付，不得作为已验证事实写入最终结论",
    }


__all__ = [
    "DEFAULT_EXPECTED_DELIVERABLE",
    "FAIL_BELOW_MIN_LENGTH",
    "FAIL_DUPLICATE_CONTENT",
    "FAIL_EMPTY_CONTENT",
    "FAIL_MISSING_CONCLUSION",
    "FAIL_MISSING_EVIDENCE",
    "FAIL_SCHEMA_INVALID",
    "FAIL_UNVERIFIED_EVIDENCE_REF",
    "InvalidTaskContractError",
    "MafFallbackPolicy",
    "MafManagerReviewItem",
    "MafManagerReviewOutput",
    "MafReviewDecision",
    "MafSeatDelivery",
    "MafTaskContract",
    "OUTCOME_ACCEPTED",
    "OUTCOME_ASK_USER",
    "OUTCOME_CONTROLLED_GAP",
    "OUTCOME_REASSIGNED",
    "OUTCOME_REWORK",
    "TASK_ACCEPTED",
    "TASK_ALL_STATUSES",
    "TASK_ASSIGNED",
    "TASK_CONTROLLED_GAP",
    "TASK_DELIVERING",
    "TASK_EXHAUSTED",
    "TASK_MANAGER_REVIEW",
    "TASK_REASSIGNED",
    "TASK_REWORK",
    "TASK_SUBMITTED",
    "TASK_TERMINAL_STATUSES",
    "build_manager_review_prompt",
    "build_task_contract",
    "check_delivery_deterministic",
    "conclusion_digest",
    "delivery_from_seat_output",
    "finalize_review_decision",
    "make_controlled_gap",
    "parse_manager_review",
    "repair_requirements_for_codes",
]