feat(handoff): complete context and strict receiver independent of review packets

docs/architecture/rfcs/capable-manager-semantic-handoff-v0.md

@@ -356,6 +356,10 @@ The same contract works when a research worker asks another worker to counterche
 
 | Current seam | Target | Retirement condition |
 | --- | --- | --- |
+| `review_packet.py` context assembly | `handoff/project_agent_context.py`, shared by full packet and direct handoff-only | Shared source assembly is implemented; duplicate packet-owned derivation removed |
+| `review_packet.py` human judgment, gate display and rendering | Retained presentation adapter | Owns no generic handoff state or new authority |
+| Handoff length control, fragmentation and reassembly | Channel codec plus `handoff restore` receiver | Real producer/receiver CLI validates complete fields, strict errors and in-budget compatibility; no Lark/cross-host qualification |
+| Handoff request identity, assessment, result and recovery | M2/M3 collaboration owner | Separately qualified; codec digests are neither request identity nor ownership receipts |
 | Manager inherits Chat planning-only restrictions and JSON preview fallback | Dedicated capable-manager role using native host tools and accepted effect receipts; keep explicit plan-only mode for users who select it | M1 proves ordinary authorized actions and restricted-mode parity; remove contradictory manager instructions |
 | Manager context inbox plus same-Goal Todo-handoff rules | One collaboration work-request contract with referenced semantic context and intent-specific admission | M2 lossless migration and two-consumer conformance; remove duplicate identities and transitions |
 | `manager_context` owns generic dispatch/decision semantics in Python | Core TypeScript collaboration domain; Python calls the typed boundary and adapts runtime/channel I/O | Switch one writer after differential tests, then delete the old decision implementation |

docs/architecture/rfcs/capable-manager-semantic-handoff-v0.zh-CN.md

@@ -331,6 +331,10 @@ provider 缺失、索引过期/不完整、超时、零命中，保留 typed gap
 
 | 当前边界 | 目标 | 何时删除旧路径 |
 | --- | --- | --- |
+| `review_packet.py` 交接上下文组装 | `handoff/project_agent_context.py`，full packet 与直接 handoff-only 共用 | 共享来源组装已实现，删除 packet 内重复推导 |
+| `review_packet.py` 人工判断、gate 展示及渲染 | 保留为展示适配器 | 不拥有通用交接状态，不额外授予权限 |
+| 交接长度控制、分片与恢复 | 渠道 codec 与 `handoff restore` 接收入口 | 真实生产→接收 CLI 验证完整字段、严格错误及预算内兼容；不宣称飞书或跨主机验收 |
+| 交接请求身份、评估、结果及恢复关系 | M2/M3 collaboration owner | 独立验收；codec digest 不是请求身份或所有权回执 |
 | 管家继承 Chat planning-only 限制和 JSON 预览兜底 | 独立强能力管家角色，使用原生工具和已接受 effect 回执；用户主动选择时保留 plan-only 模式 | M1 验证普通授权操作、受限模式，再删矛盾指令 |
 | 管家上下文 inbox 与 same-Goal Todo-handoff 规则并存 | 一个工作请求契约，引用语义背景，按具体意图检查准入 | M2 无损迁移、双消费者验证后，删重复身份与转移 |
 | `manager_context` 在 Python 掌握通用 dispatch/decision 语义 | Core TS collaboration domain；Python 只调用 typed 边界、适配 runtime/通道 I/O | 差分验证后切单 writer，再删旧判断实现 |

docs/status-data-contract.md

@@ -1660,11 +1660,14 @@ Review Packet source-of-truth rule:
   receives a small current instruction;
 - `loopx review-packet --goal-id <goal-id> --handoff-only` is the
   copy-minimal form for an already selected or approved target-agent relay: it
-  prints only the `project_agent_handoff` text in markdown output, while JSON
+  prints only agent context (or its complete shard set) in markdown output, while JSON
   output returns a minimized handoff payload instead of the full operator
   packet. To keep the hot path compact, handoff-only JSON does not expose a separate
   `handoff_followthrough_summary` prose field; that prose remains available in
-  the full Review Packet and embedded handoff text;
+  the full Review Packet and embedded handoff text. A fragmented handoff is the
+  one case where handoff-only output exceeds one shard: markdown prints every
+  shard with a relay header and JSON adds `project_agent_handoff_fragments` plus
+  `handoff_fragment_manifest`, without adding any other Review Packet content;
 - project-agent handoff commands redact local absolute registry/runtime paths
   before they enter `project_agent_command`, `project_agent_handoff`, or
   `handoff_text`;
@@ -1673,6 +1676,48 @@ Review Packet source-of-truth rule:
   block, and carry only the target goal guard, minimal-context rule, source
   label, optional compact post-handoff delivery scale, optional delivery
   contract, forwarding/execution boundary, command, and stop condition;
+- overflow preserves the **prepared handoff text**, after existing command-block
+  normalization and bounded status projection. This is not a promise to preserve
+  raw source documents or the original multiline command spelling. No sections
+  are deleted to fit the transport budget. Complete `project_agent_handoff` and
+  `handoff_text` fields always contain the entire prepared text, including on
+  overflow; consumers ignoring new keys still receive the complete instruction.
+  `handoff_interface_budget` and handoff-only size fields measure that complete
+  text and report `within_budget=false` when appropriate. The 16 line / 1800
+  character limit is a **per-shard** transport budget, not a total semantic cap;
+- on overflow, `project_agent_handoff_fragments` contains **all ordered shards**,
+  including index 0, with `handoff_fragment_manifest` describing the set and
+  original/per-shard sizes. Markdown renders that set once. On the in-budget
+  path no extra keys or envelopes are added, preserving existing output.
+  Each shard carries a `<!--loopx-handoff ... -->` envelope with content-derived
+  set id, sequence, payload checksum, previous-shard hash and full-content digest.
+  Fences are closed/reopened and long lines continued using reserved transport
+  markers. Reassembly restores prepared text byte-for-byte;
+- receivers use `loopx handoff restore --input handoff.json --format json` for
+  full or handoff-only producer JSON, or add `--input-format markdown` for raw
+  sharded Markdown (full packet or handoff-only). `--input -` reads stdin.
+  Unfragmented Markdown must be handoff-only; unfragmented full packets should
+  use JSON. JSON is recommended because Markdown renderers may strip comments.
+  A Markdown fragment title with a missing envelope, or an envelope moved off
+  the start of its line, fails restoration instead of becoming unverified plain
+  text. If a renderer removes both titles and envelopes, the remaining text
+  cannot be identified as fragmented; obtain the original JSON output.
+  Plain unframed text has no integrity proof. Reserved fragment titles,
+  envelopes, continuation and fence markers cannot be supplied as oversized
+  source content;
+- restoration strictly rejects missing, reordered, duplicate, mixed-set or
+  modified shards, malformed envelopes, inconsistent complete text fields and
+  mismatched manifests. Failures exit nonzero, expose an `error_code`, and return
+  no partial `handoff_text`. Collect all parts in order and retry with unchanged
+  producer output. There is no incremental/idempotent collector: repeated
+  generation is deterministic, but duplicate parts in one import are errors;
+- checksums prove content consistency, not sender authentication, request
+  identity, receiver acceptance or execution authority. Equal text from two
+  requests must not be business-deduplicated by set id. Restore never executes
+  content, changes Todo/claim/lease, starts a session or opens the registry.
+  Recheck current goal, scope and applicable gates after restoring; use existing
+  `handoff prepare/inspect/adopt` for ownership where applicable. This CLI path
+  does not qualify Lark delivery, cross-host recovery or arbitrary renderers;
 - `handoff_delivery_contract` is optional structured guidance derived from the
   current `handoff_readiness` plus `project_asset.execution_profile`, not a
   target-specific hack. When repeated small-scale follow-through reaches the

examples/control_plane/review-packet-cli-smoke.py

@@ -750,6 +750,35 @@ def assert_dense_handoff_stays_within_budget() -> None:
     assert_handoff_only_top_level_budget(handoff_only, "dense handoff-only json")
 
 
+def assert_overflow_cli_roundtrip() -> None:
+    """The real producer preserves constraints and the receiver cannot execute them."""
+    with tempfile.TemporaryDirectory(prefix="loopx-handoff-overflow-") as tmp:
+        root = Path(tmp)
+        registry = write_planned_registry(root)
+        mark_owner_review_todo_done(root)
+        # An independent synthetic approved command makes the actual CLI overflow.
+        command = "printf '%s' '" + "preserve-source-evidence " * 100 + "RETURN-VALIDATION'"
+        append_operator_gate_approval_fixture(root, command=command)
+        for handoff_only in (False, True):
+            flags = ["--handoff-only"] if handoff_only else []
+            args = ["review-packet", "--goal-id", GOAL_ID, "--scan-root", str(root / "project"), *flags]
+            payload = json.loads(run_cli(root, registry, "--format", "json", *args).stdout)
+            complete = payload["project_agent_handoff"]
+            assert payload["handoff_interface_budget"]["within_budget"] is False
+            assert "authority/material: topics=2, materials=4" in complete
+            assert "Run the read-only map dry-run after owner todo resolution." in complete
+            assert "生产动作、更高权限" in complete
+            assert command in complete
+            for input_format in ("json", "markdown"):
+                source = root / f"received.{input_format}"
+                source.write_text(run_cli(root, registry, "--format", input_format, *args).stdout)
+                result = run_cli(root, registry, "--format", "json", "handoff", "restore",
+                                 "--input", str(source), "--input-format", input_format)
+                restored = json.loads(result.stdout)
+                assert restored == {"ok": True, "handoff_text": complete}, restored
+                assert "【人只需判断】" not in restored["handoff_text"]
+
+
 def main() -> int:
     help_result = subprocess.run(
         [sys.executable, "-m", "loopx.cli", "review-packet", "--help"],
@@ -762,6 +791,7 @@ def main() -> int:
     assert "JSON output returns a minimized handoff payload" in compact_help, help_result.stdout
     assert "JSON output keeps the full payload" not in compact_help, help_result.stdout
 
+    assert_overflow_cli_roundtrip()
     assert_status_data_contract_documents_handoff_budget()
     assert_attention_queue_drives_approved_handoff_over_stale_history()
     assert_project_agent_handoff_prioritizes_advancement_todos()

examples/control_plane/review_packet_cli_fixtures.py

@@ -6,7 +6,6 @@
 import re
 import subprocess
 import sys
-import tempfile
 from collections.abc import Iterator
 from datetime import datetime, timezone
 from pathlib import Path
@@ -197,7 +196,7 @@ def approved_command_with_local_paths(root: Path) -> str:
     )
 
 
-def append_operator_gate_approval_fixture(root: Path) -> None:
+def append_operator_gate_approval_fixture(root: Path, *, command: str | None = None) -> None:
     run_dir = root / "runtime" / "goals" / GOAL_ID / "runs"
     run_dir.mkdir(parents=True, exist_ok=True)
     generated_at = datetime.now(timezone.utc).replace(microsecond=0).isoformat()
@@ -216,7 +215,7 @@ def append_operator_gate_approval_fixture(root: Path) -> None:
             "decision": "approve",
             "operator_question": f"是否同意 `{GOAL_ID}` 先执行 read-only map opt-in？",
             "reason_summary": f"同意 {GOAL_ID} 先做 read-only map dry-run，不授权写入或生产动作",
-            "agent_command": approved_command_with_local_paths(root),
+            "agent_command": command or approved_command_with_local_paths(root),
         },
     }
     json_path.write_text(json.dumps(record, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")

loopx/cli_commands/handoff_restore.py

@@ -0,0 +1,127 @@
+"""Content-only receive adapter. Never opens a registry or executes commands."""
+
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+from typing import Any
+
+from ..control_plane.handoff.handoff_fragments import (
+    ENVELOPE_PREFIX,
+    HandoffShardError,
+    build_handoff_shard_manifest,
+    reassemble_handoff_shards,
+    restore_handoff_text,
+)
+
+
+def restore_handoff_input(text: str, *, input_format: str) -> str:
+    if input_format == "markdown":
+        if not text.strip():
+            raise HandoffShardError("missing", "empty handoff input")
+        if text.startswith("【LoopX Review Packet】") and ENVELOPE_PREFIX not in text:
+            raise HandoffShardError(
+                "input",
+                "use full packet JSON or handoff-only Markdown for an unfragmented packet",
+            )
+        # Unframed input remains content; apparent fragments must verify first.
+        return restore_handoff_text(text)
+    try:
+        value = json.loads(text)
+    except json.JSONDecodeError as exc:
+        raise HandoffShardError(
+            "input", "invalid JSON; select --input-format markdown for raw Markdown"
+        ) from exc
+    if not isinstance(value, dict) or value.get("ok") is not True:
+        raise HandoffShardError(
+            "input", "expected a successful review-packet JSON object"
+        )
+    shards = value.get("project_agent_handoff_fragments")
+    manifest = value.get("handoff_fragment_manifest")
+    fields = [
+        value[key] for key in ("project_agent_handoff", "handoff_text") if key in value
+    ]
+    if not fields or any(not isinstance(field, str) or not field for field in fields):
+        raise HandoffShardError("input", "missing complete handoff text field")
+    if shards is not None or manifest is not None:
+        if not isinstance(shards, list) or not shards:
+            raise HandoffShardError(
+                "missing",
+                "missing handoff fragments; obtain the complete producer output",
+            )
+        if any(not isinstance(shard, str) for shard in shards):
+            raise HandoffShardError("input", "handoff fragments must be strings")
+        restored = reassemble_handoff_shards(shards)
+        if not isinstance(manifest, dict) or manifest != build_handoff_shard_manifest(
+            restored, shards
+        ):
+            raise HandoffShardError(
+                "manifest", "manifest does not match the verified fragment set"
+            )
+    else:
+        restored = restore_handoff_text(fields[0])
+    if any(field != restored for field in fields):
+        raise HandoffShardError(
+            "integrity", "complete handoff fields disagree with the recovered text"
+        )
+    return restored
+
+
+def handle_handoff_restore(args: Any, *, output_format: Any, print_payload: Any) -> int:
+    try:
+        # Reject ownership arguments rather than implying restoration adopts work.
+        ownership_fields = (
+            "goal_id",
+            "todo_id",
+            "agent_id",
+            "session_id",
+            "operation_id",
+            "expected_revision",
+            "rationale",
+            "source_ref",
+            "artifact",
+            "target_agent_id",
+            "task_lease_idempotency_key",
+            "task_lease_expected_version",
+            "from_context",
+        )
+        if any(
+            getattr(args, key, None) is not None and getattr(args, key, None) != []
+            for key in ownership_fields
+        ):
+            raise HandoffShardError(
+                "input",
+                "restore accepts content only; ownership arguments belong to prepare/inspect/adopt",
+            )
+        if args.handoff_format == "digest":
+            raise HandoffShardError(
+                "input", "restore supports --format json or markdown"
+            )
+        if not args.input:
+            raise HandoffShardError(
+                "input", "restore requires --input FILE (or - for stdin)"
+            )
+        text = (
+            sys.stdin.read()
+            if args.input == "-"
+            else Path(args.input).read_text(encoding="utf-8")
+        )
+        restored = restore_handoff_input(text, input_format=args.input_format)
+        payload: dict[str, Any] = {"ok": True, "handoff_text": restored}
+    except (HandoffShardError, OSError, UnicodeError) as exc:
+        payload = {
+            "ok": False,
+            "error_code": getattr(exc, "code", "input"),
+            "error": str(exc),
+            "next_action": "Obtain the unchanged complete handoff and retry; no content was executed or adopted.",
+        }
+    fmt = args.handoff_format or output_format(args)
+    if payload["ok"] and fmt == "markdown":
+        # Do not add a newline to byte-exact decoded content.
+        sys.stdout.write(payload["handoff_text"])
+    else:
+        print_payload(
+            payload, fmt, lambda value: json.dumps(value, ensure_ascii=False, indent=2)
+        )
+    return 0 if payload["ok"] else 1

loopx/cli_commands/status.py

@@ -25,10 +25,15 @@
     compact_agent_lane_todo_index_for_status_display,
 )
 from ..diagnose import collect_diagnosis, render_diagnosis_markdown
+from ..control_plane.handoff.project_agent_context import build_project_agent_handoff
+from ..control_plane.handoff.handoff_fragments import render_handoff_transport
 from ..handoff_budget import build_handoff_interface_budget
 from ..presentation.renderers.status_markdown import render_status_markdown
 from ..quota import build_quota_should_run
-from ..review_packet import build_review_packet, render_review_packet_markdown
+from ..review_packet import (
+    build_review_packet,
+    render_review_packet_markdown,
+)
 from ..status import AUTONOMOUS_REPLAN_PERIODIC_LOOKBACK, collect_status
 from .status_registration import register_status_commands as register_status_commands
 
@@ -136,6 +141,12 @@ def review_packet_handoff_only_payload(payload: dict[str, object]) -> dict[str,
             "within_budget": handoff_budget.get("within_budget"),
         }
     )
+    fragment_texts = payload.get("project_agent_handoff_fragments")
+    if isinstance(fragment_texts, list) and fragment_texts:
+        result["project_agent_handoff_fragments"] = fragment_texts
+        result["handoff_fragment_manifest"] = payload.get(
+            "handoff_fragment_manifest"
+        )
     return result
 
 
@@ -876,12 +887,15 @@ def handle_review_packet_command(
         )
         if args.agent_id:
             attach_agent_lane_next_actions(status_payload, agent_id=args.agent_id)
-        payload = build_review_packet(
-            status_payload,
-            goal_id=args.goal_id,
-            action_kind=args.action_kind,
-            review_url=args.review_url,
-        )
+        if args.handoff_only:
+            payload = build_project_agent_handoff(
+                status_payload, goal_id=args.goal_id, action_kind=args.action_kind,
+            )
+        else:
+            payload = build_review_packet(
+                status_payload, goal_id=args.goal_id, action_kind=args.action_kind,
+                review_url=args.review_url,
+            )
     except Exception as exc:
         payload = {
             "ok": False,
@@ -891,7 +905,14 @@ def handle_review_packet_command(
     if args.handoff_only:
         payload = review_packet_handoff_only_payload(payload)
     if args.handoff_only and selected_format != "json" and payload.get("ok"):
-        print(str(payload.get("handoff_text") or ""))
+        fragment_texts = payload.get("project_agent_handoff_fragments")
+        if not isinstance(fragment_texts, list):
+            fragment_texts = []
+        print(
+            render_handoff_transport(
+                str(payload.get("handoff_text") or ""), fragment_texts
+            )
+        )
     else:
         print_payload(payload, selected_format, render_review_packet_markdown)
     return 0 if payload.get("ok") else 1