项目初始化保留同会话已有事实；不是每次只从最新一句话重来。

B工作树 · 所引文件无未提交改动 · HEAD edbc7dad958b；均不自动代表线上版本。

def _initialize_locked(team_id: str, objective: str, plan: dict, *, mode: str,
                       user_id: str = "", conversation_id: str = "") -> dict:
    """调用者已持有 _packet_lock；合并原有项目事实，不另建执行现场。"""
    source = plan.get("project_context") if isinstance(plan.get("project_context"), dict) else {}
    packet = _json_safe(source) if source else {}
    previous = load(team_id)
    conversation = str(conversation_id or packet.get("conversation_id") or
                       previous.get("conversation_id") or team_id)
    require_owner(previous, user_id, conversation)
    # 显式调用沿本次身份表示传给 workdir；未指定用户的维护调用保留原归属。
    owner = str(user_id or previous.get("user_id") or packet.get("user_id") or "")
    same_context = bool(previous and str(previous.get("conversation_id") or "") == conversation)
    if same_context:
        for key in ("decisions", "completed", "open_gaps", "verification_findings", "artifacts", "attempts",
                    "source_ref", "input_snapshot", "version_identity", "execution_role", "quality_context",
                    "visibility", "reference_atlas", "frontier_research_contract"):
            if key not in packet and previous.get(key) not in (None, "", [], {}):
                packet[key] = _json_safe(previous[key])
        # 续做计划可能只携带任务状态，工程现场仍以该会话的持久 packet 为准。
        # 两个路径别名成组继承：显式给新 workspace 时不能再补回旧 repo_root。
        if not any(key in packet for key in ("repo_root", "workspace_root")):
            for key in ("repo_root", "workspace_root"):
                if previous.get(key):
                    packet[key] = _json_safe(previous[key])

文件 SHA-256：e22fb78254e2e7fdde771a8be35fbef1e727542f544bf00ca12e75cb3c789c68

当前主链按任务取项目快照，经project_context参数交给实际执行入口。

B工作树 · 含本地未提交改动 · HEAD edbc7dad958b；均不自动代表线上版本。

        try:
            from legion_contracts import CapabilityPolicy, TaskFacts
            _node_spec = raw_tasks.get(k) or {} if isinstance(raw_tasks.get(k), dict) else {}
            _verification_role = str(_node_spec.get("verification_role") or "")
            _verification_criteria = list(_node_spec.get("verification_criteria") or [])
            _project_snapshot = (_project_context.snapshot(team_id, k, packet=_project_packet)
                                 if _project_context else dict(_project_packet or {}))
            _task_facts = TaskFacts(
                goal=question, role=role_task or "回答问题", task_id=k, team_id=team_id,
                attempt_id=int(attempt or 0), input_hash=str(_node_spec.get("input_hash") or ""),
                idempotency_key=str(_node_spec.get("idempotency_key") or ""),
                delivery_contract=dict(plan.get("delivery_contract") or {}),
                orchestration={**(orchestration_spec.to_dict() if orchestration_spec else {}),
                               "assignment": {"task_id": k, "harness": harness_key,
                                              "source": _node_spec.get("assignment_source"),
                                              "native_capabilities": _node_spec.get("native_capabilities") or {}}},
                deliverables=list((plan.get("deliverables") or [])),
                acceptance=list((plan.get("acceptance") or [])) if isinstance(plan.get("acceptance"), list)
                           else ([str(plan.get("acceptance"))] if plan.get("acceptance") else []),
                upstream_artifacts=_upstream_refs,
                upstream_results=[{"key": d, "status": (results.get(d) or {}).get("status"),
                                   "error_envelope": (results.get(d) or {}).get("error_envelope")}
                                  for d in deps],
                project_context=_project_snapshot,
            ).to_dict()

文件 SHA-256：d3f0cd8057e71357cfe3449caccd7c09164ebf869a6dcae50079da64265a683d

原生上下文装配有PROJECT_CONTEXT_JSON消费者；收到不等于理解或采用。

B工作树 · 所引文件无未提交改动 · HEAD edbc7dad958b；均不自动代表线上版本。

    # project_context 有独立高权重 section；从 task_facts 删除同名副本，避免同一项目现场双写计费。
    prompt_facts = dict(ctx.task_facts or {})
    prompt_facts.pop("project_context", None)
    # 文件起点索引只供运行时观察；不改变模型看到的项目语义，也不产生无项目的伪项目块。
    prompt_project = {key: value for key, value in (ctx.project_context or {}).items()
                      if key != "_artifact_input_baseline"}
    if quick and prompt_project:
        for duplicate in ("goal", "team_id", "delivery_contract", "acceptance", "deliverables"):
            prompt_facts.pop(duplicate, None)
        prompt_facts = {key: value for key, value in prompt_facts.items()
                        if value not in (None, "", [], {}, 0)}
    facts_block = ("\n\nTASK_FACTS_JSON\n" + _json.dumps(prompt_facts, ensure_ascii=False, separators=(",", ":"))
                   if prompt_facts else "")
    policy_block = ("\n\nCAPABILITY_POLICY_JSON\n" + _json.dumps(ctx.capability_policy, ensure_ascii=False, separators=(",", ":"))
                    if ctx.capability_policy and not quick else "")
    memory_block = ("\n\nTASK_MEMORY_JSON\n" + _json.dumps(ctx.task_memory, ensure_ascii=False, separators=(",", ":"))
                    if ctx.task_memory else "")
    project_block = ("\n\nPROJECT_CONTEXT_JSON\n" +
                     _json.dumps(prompt_project, ensure_ascii=False, separators=(",", ":"))
                     if prompt_project else "")

文件 SHA-256：336aea7d8259df32058fe89390ed8cbcac94c4eba99775a87a6a00632cb384f1

原生传输使用有界投影与原文引用，保留完整恢复记录；不是无限上下文。

B工作树 · 所引文件无未提交改动 · HEAD edbc7dad958b；均不自动代表线上版本。

def native_context(team_id: str, key: str) -> dict:
    """仅给原生执行入口的读取投影；恢复/UI仍使用完整context和原记录。"""
    with _LOCK:
        view = context(team_id, key)
        state = load(team_id, key)
        if not view or state.get("team_id") != team_id or state.get("key") != key:
            return view
        path = _path(team_id, key)
        if path.is_symlink() or path.parent.is_symlink() or ROOT.is_symlink():
            return view
        view = dict(view)
        import project_context
        if isinstance(view.get("artifacts"), list):
            view["artifacts"] = project_context._snapshot_artifacts(
                team_id, state, view["artifacts"], source_path=path)
        # 完整原生历史和工具事实仍在本请求原节点；只省传输，不删事实，
        # 不改变native checkpoint、工作目录或失败状态。
        nodes = {"/" + name: view[name] for name in ("completed_steps", "native_facts")
                 if name in view}
        refs = project_context._snapshot_references(path, state, nodes)
        for name in ("completed_steps", "native_facts"):
            if "/" + name in refs:
                view.pop(name, None)
                view[name + "_ref"] = refs["/" + name]
        return view

文件 SHA-256：f4bf200018f519a44ac376df3f6dc51c69081ea8e59d60ed6dfcc7be8f9f29e4

只有真实多语向量模式才进入对应混排，不能把所有回退都称语义召回。

B工作树 · 所引文件无未提交改动 · HEAD edbc7dad958b；均不自动代表线上版本。

    # 字符 n-gram 不是跨语言语义：中英零交集时仍会因泛中文描述产生噪声。只有真正的多语言
    # embedding 才参与混排；否则诚实保留 BM25，交给 harness 读 manifest 后自行改写重搜。
    semantic_weight = 0.55 if semantic_backend == "multilingual_embedding" else 0.0
    lexical_weight = 1.0 - semantic_weight
    for item in catalog:
        score = (lexical_weight * lexical.get(item["id"], 0.0) / lex_max
                 + semantic_weight * max(0.0, semantic.get(item["id"], 0.0)))

文件 SHA-256：d85583d5b124599bd440ccc19ecc99cb91d5b1795f56bf97698f3dee702b5507

向量召回依赖开关与完整本地模型条件；源码存在不代表当前启用。

B工作树 · 所引文件无未提交改动 · HEAD edbc7dad958b；均不自动代表线上版本。

    if os.environ.get("LEGION_TOOL_LOCAL_EMBEDDING", "0") != "1":
        return None
    return _embedding_scores(items, query)


def _embedding_scores(items: list[dict], query: str) -> list[float] | None:
    # 下载/修复 embedding 模型属于显式安装动作，绝不能在每次工具检索时偷偷联网重试。
    # 只有本地快照完整且缓存索引已存在时才启用；否则立即回退字符 n-gram。
    model_cache = Path.home() / ".cache" / "huggingface" / "hub" / \
        "models--sentence-transformers--paraphrase-multilingual-MiniLM-L12-v2"
    snapshots = model_cache / "snapshots"
    required = ("modules.json", "config.json", "tokenizer.json", "model.safetensors")
    complete_snapshot = next((p for p in snapshots.glob("*")
                              if p.is_dir() and all((p / name).exists() for name in required)), None) \
        if snapshots.is_dir() else None
    if not CACHE.is_file() or complete_snapshot is None:
        return None

文件 SHA-256：1ea6d92a16ca85cbb2cba80beca096a6c01c843b4a2f97b16faac6358ce5f6ce

没有候选时仍可走字符片段正分回退；这是回退，不是跨语言语义证明。

B工作树 · 所引文件无未提交改动 · HEAD edbc7dad958b；均不自动代表线上版本。

    fallback_semantic = False
    # 模型改写不可用/超时时，通用字符向量只作为“避免交白卷”的最后候选源；仅收正分项，
    # 不随机塞零相关工具，也不覆盖已有 BM25 候选。
    if not supply_pool and semantic_backend == "character_ngram_fallback":
        semantic_ranked = sorted(
            ((score, item) for item, score in zip(catalog, sem_scores) if score > 0),
            key=lambda pair: -pair[0])
        supply_pool = [item for _score, item in semantic_ranked[:max(top * 3, top)]]
        hits = supply_pool[:top]
        fallback_semantic = bool(supply_pool)

文件 SHA-256：d85583d5b124599bd440ccc19ecc99cb91d5b1795f56bf97698f3dee702b5507

可执行能力调用要核依赖并落到真实函数；参考源码不自动成为工具。

B工作树 · 所引文件无未提交改动 · HEAD edbc7dad958b；均不自动代表线上版本。

                names.add(n[2:])
    return sorted(names)


def call(tool_name: str, path: str = "", question: str = "", timeout_s: int = 120) -> dict:
    """无状态桥调 auto 的 t_<tool_name>(跨 capability_stack+qa_tool_impls)·stub session·绕开 auto pipeline.

    path: 本地文件路径(OCR/PDF/图像类解析它);question: 覆盖式问题(优先于path).
    返回 {ok, result, tool, via:'capstack_bridge'}。核心工具无状态·session 用空 dict 承接记录。
    """
    fn = _find_fn(tool_name)
    if not callable(fn):
        return {"ok": False, "runnable": False,
                "error": f"工具源码仅供参考或无状态桥/随包依赖未就绪: t_{tool_name}",
                "hint": f"可桥调的: {list_bridged_tools()[:20]}"}
    # 构造 question(工具内 _parse_local_path 从 question 提路径)+ stub session(空dict即可)
    q = question or (path or "")
    stub_session: dict = {"extracted_values": [], "_budget": {}}
    try:
        result = fn(q, stub_session)
        return {"ok": True, "result": str(result), "tool": tool_name,
                "via": "capstack_bridge",
                "extracted_values": stub_session.get("extracted_values", [])}
    except Exception as e:
        return {"ok": False, "error": f"桥调 t_{tool_name} 失败: {type(e).__name__}: {e}",

文件 SHA-256：bcab53ce886b2ee04c5552b192781f19c1fa70cd70473fd8637859eeeffbe238

自动发现先形成候选描述，不把任意GitHub目录直接视为可运行成员。

B工作树 · 所引文件无未提交改动 · HEAD edbc7dad958b；均不自动代表线上版本。

def discover_local(path: str, url: str = "", ref: str = "HEAD") -> dict:
    """读取项目自述元数据形成候选 manifest；不自动执行、不猜未声明协议。"""
    root = pathlib.Path(path).expanduser().resolve()
    if not root.is_dir():
        return {"ok": False, "error": f"目录不存在:{root}"}
    package = _read_json(root / "package.json")
    pyproject = _read_text(root / "pyproject.toml")
    cargo = _read_text(root / "Cargo.toml")
    readme = _read_text(root / "README.md", 80_000)
    project_name = ""
    if pyproject:
        match = re.search(r"(?m)^\s*name\s*=\s*[\"']([^\"']+)[\"']", pyproject)
        project_name = match.group(1) if match else ""
    name = package.get("name") or project_name or root.name
    key = re.sub(r"[^a-z0-9_.-]+", "-", str(name).lower()).strip("-.")
    runtime: dict[str, Any] = {}
    install: list[list[str]] = []

文件 SHA-256：a82e04c32d3ce5edfeaaf92f655349946f1bd4d1b764e56577685a2d9079040c

注册先校验manifest并以锁更新目录，让接入有统一入口。

B工作树 · 所引文件无未提交改动 · HEAD edbc7dad958b；均不自动代表线上版本。

def _upsert_catalog(key, spec_path):
    entry = json.loads(pathlib.Path(spec_path).read_text(encoding="utf-8"))
    if entry.get("key") != key:
        raise ValueError("spec.key=%s 与 --key=%s 不一致" % (entry.get("key"), key))
    validation = R.agent_manifest.validate(entry)
    if not validation.get("ok"):
        raise ValueError("spec 不满足 catalog 契约:" + ";".join(validation.get("errors") or []))
    with R.catalog_lock():
        items = R.load_catalog()
        replaced = False
        for i, x in enumerate(items):
            if x.get("key") == key:
                items[i] = entry
                replaced = True
                break
        if not replaced:
            items.append(entry)
        R._save_catalog(items)
    return "catalog %s:%s (total=%d)" % ("updated" if replaced else "appended", key, len(items))

文件 SHA-256：7b015abc9dedb6591c0af2f0f2c1c5e6d707a97424bb050bd93822c80052a812

可用判断区分版本、探活、凭据、协议和角色；在目录里不等于可执行。

B工作树 · 所引文件无未提交改动 · HEAD edbc7dad958b；均不自动代表线上版本。

    credentials = _credentials(item, env)
    has_transport = bool(item.get("transports"))
    executable_role = item.get("execution_role") not in ("control-plane", "ui-platform", "capability")
    execution_ready = bool(revision_ok and available and credentials["ready"] and has_transport and executable_role)

文件 SHA-256：5250fcb3d12397047cf4ce3d4c9663147d237afa13f7d40274e24bf3a8de74ac

现行总PASS只排除FAIL，SKIP仍可能存在；必须看分步记录。

B工作树 · 所引文件无未提交改动 · HEAD edbc7dad958b；均不自动代表线上版本。

    passed = sum(1 for x in steps if x.status == "PASS")
    failed = sum(1 for x in steps if x.status == "FAIL")
    skipped = sum(1 for x in steps if x.status == "SKIP")
    verdict = "PASS" if failed == 0 else "FAIL"

文件 SHA-256：7b015abc9dedb6591c0af2f0f2c1c5e6d707a97424bb050bd93822c80052a812

后续任务可检索旧经验；召回是输入事实，不是采用或正确性证明。

B工作树 · 含本地未提交改动 · HEAD edbc7dad958b；均不自动代表线上版本。

    try:
        import legion_memory
        # user_id 隔离召回(军团自审 P0):只捞本用户归属的经验+全局共享经验(working/procedural),
        # 不会把别的用户的任务产出召回给当前用户(_visible_to 按 user_id 过滤)。
        r = legion_memory.search(question, top=2, user_id=user_id or "")
    except Exception:
        return ""
    exps = r.get("experiences") if isinstance(r, dict) else None
    if not exps:
        return ""
    # 用户可见:如实陈述【发生的事实】——给军团注入了 N 条历史经验供参考。★ context not control:
    # 不替军团断言"参考了/辅助了本次回答"(用没用是军团的事·只有它自己知道·父层绝不替它宣称,
    # 那正是 adopted 要军团自己回报的原因)。只报"注入了什么·供参考",效果由军团自己决定。
    if cb:

文件 SHA-256：d3f0cd8057e71357cfe3449caccd7c09164ebf869a6dcae50079da64265a683d

召回结果进入执行前置上下文，说明存在实际消费接线。

B工作树 · 含本地未提交改动 · HEAD edbc7dad958b；均不自动代表线上版本。

            ctx += _dk_block
    except Exception:
        pass
    # 军团跨家经验召回:别家/自己历史上答过的相似问题·供参考(context not control·标明不一定对让模型自判)
    # 传 cb:召回到经验时给用户 emit 一条可见事件(过程透明·让"军团在学习变强"可感知·用户中心)。
    ctx += _recall_experience(question, cb=cb, user_id=user_id)

文件 SHA-256：d3f0cd8057e71357cfe3449caccd7c09164ebf869a6dcae50079da64265a683d

审查记录允许runtime_not_tested；源码引用核对不能替代补丁效果验证。

B工作树 · 所引文件无未提交改动 · HEAD edbc7dad958b；均不自动代表线上版本。

        "claim": str(entry.get("claim") or ""),
        "source_root": str(_resolve_source_root(str(entry["source_root"]))),
        "revision": str(checked.get("revision") or entry["revision"]),
        "file": str(entry["file"]),
        "line": int(entry["line"]),
        "quote": str(entry["quote"]),
        "runtime_evidence": entry.get("runtime_evidence") or "runtime_not_tested",
        "verifier": str(entry["verifier"]),
        "status": status,
        "improvement": entry.get("improvement") or None,
        "verification": {"path": checked["path"], "actual": checked["actual"],
                         "content_source": checked.get("content_source"),
                         "content_sha256": checked.get("content_sha256")},

文件 SHA-256：02b4386a4a34686c3aed56b13374cb7be43cf1b20de3dcd0f9f0ec7103ef72bf

体验目标编译成参照Context并保留已有合同；不是给作品颁世界级认证。

B工作树 · 含本地未提交改动 · HEAD edbc7dad958b；均不自动代表线上版本。

    research = {
        "goal": "与用户预期结果相关的参照资料和观察",
        "reference_atlas": deepcopy(REFERENCE_ATLAS),
        "experience_intent": intent,
        "golden_task": golden,
        "reference_requirements": {},
        "human_evidence_requirements": {},
        "comparison_policy": {},
    }
    compiled = {"quality_profile": quality_profile, "experience_intent": intent,
                "golden_task": golden, "north_star": north_star,
                "critical_dimensions": critical_dimensions,
                "claim_ceiling": "UNVALIDATED",
                "frontier_research_contract": research}
    # 整份显式合同及其未知字段不重写；按键是否存在保留 False/空值，而非用真值兜底。
    return {key: deepcopy(source[key] if key in source else value)
            for key, value in compiled.items()}


def augment_plan(plan: dict | None) -> dict:
    """只编译缺失的意图、参考与质量观察 Context，不改写执行计划。"""
    source = deepcopy(plan) if isinstance(plan, dict) else {}
    raw_intent = source.get("experience_intent") if isinstance(source.get("experience_intent"), dict) else {}
    if not raw_intent:
        return source
    # 原始意图、显式策略及历史事实均由调用者拥有；编译结果只补缺失的 Context。
    # 是否研究、分阶段或设评审节点由用户/规划器决定，不在这里补 DAG 或制造阻断。
    for key, value in compile_experience_contract(source).items():
        source.setdefault(key, value)

文件 SHA-256：fad7733e31b51a5b547cc9b8d121701a707cda5eab6e3541ac5d18e1e26fbaae

研究结果可进入共享quality packet；字段形成不等于真的研究或采用。

B工作树 · 所引文件无未提交改动 · HEAD edbc7dad958b；均不自动代表线上版本。

def _merge_frontier_context(packet: dict, context: dict, task_id: str) -> None:
    """Persist research authored by the Legion as shared context for downstream nodes."""
    if not context:
        return
    research = dict(packet.get("frontier_context") or {})
    research.update(_json_safe(context))
    research.update({"source_task_id": str(task_id), "updated_at": time.time()})
    packet["frontier_context"] = research
    # 概念锚点表(重放#3/#4实锤:研究浅/锚点不结构化→全链自洽地平庸):研究在
    # LEGION_FRONTIER_CONTEXT 里给出 concept_anchors 数组时,持久化为全军团共同对照事实,
    # 翻译官任务书/验证契约/首脑协调全部逐锚对照这张表。
    # P0-3:每锚保留完整 schema(acceptance/evidence_requirements/artifact_refs/source_sha/
    # revision/is_crown),不再只有损压缩到5字段。
    anchors = context.get("concept_anchors") if isinstance(context.get("concept_anchors"), list) else None
    if anchors:
        rows = [row for row in (_normalize_anchor_row(item) for item in anchors) if row]
        if rows:
            _commit_anchor_rows(packet, rows, change_reason="legion_frontier_research",
                                sources=[{"channel": "protocol_block", "task_id": str(task_id),
                                          "rows": len(rows), "ingested_at": time.time()}])
    quality = dict(packet.get("quality_context") or {})
    quality["frontier_context"] = research
    quality["revision"] = max(1, int(quality.get("revision") or 0) + 1)
    quality["change_reason"] = "legion_frontier_research"
    quality["updated_at"] = time.time()
    packet["quality_context"] = quality
    packet["quality_context_revision"] = quality["revision"]

文件 SHA-256：e22fb78254e2e7fdde771a8be35fbef1e727542f544bf00ca12e75cb3c789c68

当前主链按任务取项目快照，经project_context参数交给实际执行入口。

B工作树 · 含本地未提交改动 · HEAD edbc7dad958b；均不自动代表线上版本。

        try:
            from legion_contracts import CapabilityPolicy, TaskFacts
            _node_spec = raw_tasks.get(k) or {} if isinstance(raw_tasks.get(k), dict) else {}
            _verification_role = str(_node_spec.get("verification_role") or "")
            _verification_criteria = list(_node_spec.get("verification_criteria") or [])
            _project_snapshot = (_project_context.snapshot(team_id, k, packet=_project_packet)
                                 if _project_context else dict(_project_packet or {}))
            _task_facts = TaskFacts(
                goal=question, role=role_task or "回答问题", task_id=k, team_id=team_id,
                attempt_id=int(attempt or 0), input_hash=str(_node_spec.get("input_hash") or ""),
                idempotency_key=str(_node_spec.get("idempotency_key") or ""),
                delivery_contract=dict(plan.get("delivery_contract") or {}),
                orchestration={**(orchestration_spec.to_dict() if orchestration_spec else {}),
                               "assignment": {"task_id": k, "harness": harness_key,
                                              "source": _node_spec.get("assignment_source"),
                                              "native_capabilities": _node_spec.get("native_capabilities") or {}}},
                deliverables=list((plan.get("deliverables") or [])),
                acceptance=list((plan.get("acceptance") or [])) if isinstance(plan.get("acceptance"), list)
                           else ([str(plan.get("acceptance"))] if plan.get("acceptance") else []),
                upstream_artifacts=_upstream_refs,
                upstream_results=[{"key": d, "status": (results.get(d) or {}).get("status"),
                                   "error_envelope": (results.get(d) or {}).get("error_envelope")}
                                  for d in deps],
                project_context=_project_snapshot,
            ).to_dict()

文件 SHA-256：d3f0cd8057e71357cfe3449caccd7c09164ebf869a6dcae50079da64265a683d

评审与差距可以回到项目现场；观察和作品质量结论需要各自证据。

B工作树 · 所引文件无未提交改动 · HEAD edbc7dad958b；均不自动代表线上版本。

def _record_verification_unlocked(team_id: str, report: dict) -> dict:
    packet = load(team_id)
    if not packet:
        return {}
    safe_report = _json_safe(report)
    report_id = str(report.get("report_id") or
                    f"verification:{report.get('task_id') or ''}:{report.get('role') or ''}")
    safe_report["report_id"] = report_id
    reports = [item for item in (packet.get("verification_findings") or [])
               if isinstance(item, dict) and str(item.get("report_id") or "") != report_id]
    reports.append(safe_report)
    packet["verification_findings"] = reports[-_MAX_ITEMS:]
    if str(report.get("role") or "") == "direction_reviewer":
        quality = dict(packet.get("quality_context") or {})
        quality.setdefault("north_star", _json_safe(packet.get("north_star") or {}))
        quality.setdefault("critical_dimensions", _json_safe(packet.get("critical_dimensions") or []))
        quality.setdefault("claim_ceiling", str(packet.get("claim_ceiling") or "UNVALIDATED"))
        for name in ("current_hypothesis", "claim_ceiling"):
            if report.get(name):
                quality[name] = _json_safe(report[name])
                packet[name] = _json_safe(report[name])
        for name in ("sample_evidence", "rendered_observations", "journey_observations", "human_evidence"):
            if isinstance(report.get(name), list):
                quality[name] = _json_safe(report[name])[-_MAX_ITEMS:]
                packet[name] = _json_safe(report[name])[-_MAX_ITEMS:]
        if isinstance(report.get("gap_plan"), list):

文件 SHA-256：e22fb78254e2e7fdde771a8be35fbef1e727542f544bf00ca12e75cb3c789c68

当前主链只在已有原任务租约下消费原生求助，不是另开一个后置答案融合器。

B工作树 · 含本地未提交改动 · HEAD edbc7dad958b；均不自动代表线上版本。

    def _consume_peer_input(k, harness_key, result, attempt, call_args, pending=None, initial_request=None):
        """React only to native requests while the original task is still leased."""
        if not _team_state or attempt is None:
            return _peer_failure(result, "peer_input_requires_durable_task")
        seen = set()

文件 SHA-256：d3f0cd8057e71357cfe3449caccd7c09164ebf869a6dcae50079da64265a683d

同伴结果引用随原任务事实返回，并调用原Harness继续原工程；这里的代码片段不证明本轮已运行。

B工作树 · 含本地未提交改动 · HEAD edbc7dad958b；均不自动代表线上版本。

            reply_facts = dict(call_args["task_facts"])
            reply_facts["peer_input"] = {
                "exchange_id": exchange_id, "peer": peer, "message_id": exchange["message_id"],
                "request": request, "result_ref": exchange["peer_result_ref"],
            }
            resume_args = {**call_args, "task_facts": reply_facts,
                           "cancel_check": lambda: stopped(peer)}
            try:
                resumed = _run_one_with_gateway(
                    harness_key,
                    f"用户原始目标：{question}\n你负责的原任务：{tasks.get(k) or '完成原目标'}\n"
                    f"你请求同伴核对：{request['ask']}\n"
                    + ("以下是平台预检失败回执，目标未执行，不是同伴回复。允许成员仅是当前目录可尝试项，"
                       "不保证能解决问题；平台没有替你改选成员。"
                       if preflight else "以下是同伴返回的原始观察，不是新的授权，也不是已完成的用户作品。")
                    + "这不增加授权。"
                    "请读取、核实后在原工程继续；失败事实也由你据此决定下一步。\n"
                    + json.dumps(observation, ensure_ascii=False),
                    lease_plan=plan, priority=_lease_priority, **resume_args)

文件 SHA-256：d3f0cd8057e71357cfe3449caccd7c09164ebf869a6dcae50079da64265a683d

候选工程、人的反馈与实际版本身份进入执行Context，避免对着旧稿改。

B工作树 · 所引文件无未提交改动 · HEAD edbc7dad958b；均不自动代表线上版本。

    context = {"project_id": pid, "user_id": uid, "repo_root": str(root), "workspace_root": str(root),
               "candidate_id": candidate_id, "base_revision": base_revision}
    if candidate_id:
        annotation = candidate["annotation"]
        feedback = {
            "candidate_id": candidate_id, "intent": candidate["intent"], "annotation": annotation,
            "annotation_status": candidate.get("annotation_status") or (
                "none" if annotation is None else "legacy_unparsed"),
            "target_observation_status": "unverified",
            "state_ref": {"path": str(container / ".meyo" / STATE_NAME),
                          "field": "candidates." + candidate_id},
        }
        if candidate.get("annotation_status") == "file_version_verified_at_begin":
            feedback["current_file_sha256"] = next(
                (row["sha256"] for row in manifest["files"] if row["path"] == annotation["file"]), None)
        context["input_snapshot"] = {"artifact_feedback": feedback}
        context["version_identity"] = {"project_revision": {
            **{key: candidate[key] for key in
               ("project_id", "candidate_id", "base_revision", "base_snapshot", "base_workspace_digest")},
            "workspace_digest": _digest(manifest),
        }}

文件 SHA-256：c269663046d7382159103b45c00b462d039294a6baf9266035767a336fe7ef80

采用和恢复使用同一切换路径，并携带期待版本和工作区指纹；不是质量自动判决。

B工作树 · 所引文件无未提交改动 · HEAD edbc7dad958b；均不自动代表线上版本。

def adopt(uid: str, pid: str, *, candidate_id: str, candidate_revision: str,
          expected_revision, expected_workspace_digest: str, request_id: str,
          expected_source_id: str | None = None) -> dict:
    return _switch(uid, pid, action="adopt", revision=candidate_revision, candidate_id=candidate_id,
                   expected_revision=expected_revision, expected_workspace_digest=expected_workspace_digest,
                   request_id=request_id, expected_source_id=expected_source_id)


@_api
def restore(uid: str, pid: str, *, target_revision: str, expected_revision,
            expected_workspace_digest: str, request_id: str, target_operation_id: str | None = None,
            expected_source_id: str | None = None, target_operation_side: str | None = None) -> dict:
    """'from' selects the operation's prior retained version, not its unsaved workspace."""
    return _switch(uid, pid, action="restore", revision=target_revision, expected_revision=expected_revision,
                   expected_workspace_digest=expected_workspace_digest, request_id=request_id,
                   target_operation_id=target_operation_id, expected_source_id=expected_source_id,
                   target_operation_side=target_operation_side)

文件 SHA-256：c269663046d7382159103b45c00b462d039294a6baf9266035767a336fe7ef80

正常HTTP入口消费采用/恢复请求并传递版本条件，连接用户操作与版本机制。

B工作树 · 所引文件无未提交改动 · HEAD edbc7dad958b；均不自动代表线上版本。

        elif action == "adopt":
            result = revisions.adopt(
                uid, project_id, candidate_id=body.get("candidate_id"),
                candidate_revision=body.get("candidate_revision"),
                expected_revision=body.get("expected_revision"),
                expected_workspace_digest=body.get("expected_workspace_digest"),
                request_id=body.get("request_id"),
                expected_source_id=body.get("expected_source_id"))
        elif action == "restore":
            result = revisions.restore(
                uid, project_id, target_revision=body.get("target_revision"),
                expected_revision=body.get("expected_revision"),
                expected_workspace_digest=body.get("expected_workspace_digest"),
                request_id=body.get("request_id"),
                expected_source_id=body.get("expected_source_id"),
                target_operation_id=body.get("target_operation_id"),
                target_operation_side=body.get("target_operation_side"))

文件 SHA-256：fae6e40ee995dd22a82cce6f1c19942b4eef746d9ead3dfe040795e75aa9e5cc

候选在真实writer结束后封存；SSE传输结束不等于写作结束，封存不代表已采用。

B工作树 · 所引文件无未提交改动 · HEAD edbc7dad958b；均不自动代表线上版本。

        nonlocal finished
        with lock:
            if finished:
                return
            finished = True
        ended = revisions.set_execution(
            uid, project_id, candidate_id, running=False, session_id=payload["session_id"])
        # A sealed candidate is an exact set of files, not a quality decision.
        # Native failure may still leave useful files; nothing is auto-adopted.
        frozen = revisions.seal(uid, project_id, candidate_id) if ended.get("ok") else ended
        if not frozen.get("ok"):
            handler.log_message("candidate snapshot not ready: %s", frozen.get("error"))

    handler._candidate_finalizer = finish
    try:
        return execute(payload)
    finally:
        # Dispatch owns a worker that may outlive SSE. That worker captures
        # finish and invokes it after its actual exit. run_one remains synchronous.
        if not handler._candidate_worker_started:
            finish()

文件 SHA-256：fae6e40ee995dd22a82cce6f1c19942b4eef746d9ead3dfe040795e75aa9e5cc

核目标commit、文件集合与逐文件SHA；该脚本可下载，研究仅阅读，没有运行。

开发主仓 · 所引文件无未提交改动 · HEAD 600b4a59f50a；均不自动代表线上版本。

assert actual == expected_files, {

文件 SHA-256：2b6da35ff3f4d33e9a1271bac04ea20a05081dc27f857a3b58324d4374b795ff

过滤规则先于递归包含；只读源码，不执行同步或删除。

开发主仓 · 所引文件无未提交改动 · HEAD 600b4a59f50a；均不自动代表线上版本。

# rsync uses the first matching rule: exclusions must precede recursive includes.

文件 SHA-256：c8a7790357e9b70d7fc7bff4952751e148e52ab76239a97f81e3d556058455e1

以缓存身份与实际入口识别是否可复用；不是此刻平台缓存命中的证明。

发布仓工作副本 · 所引文件无未提交改动 · HEAD cdd8125e09a2；均不自动代表线上版本。

  if [ -f "$stamp" ] && [ "$(cat "$stamp")" = "$pkg_hash" ] && npm_entry_valid "$prefix" "$pkg" "$bin"; then

文件 SHA-256：1c6ec802f473bc71a4bca89dc426a66e6e6e529dd0b8ad95071c6e2ddb8f6904

绑定固定源码、锁文件、解释器与安装配方；缓存命中明确不充当运行验收。

发布仓工作副本 · 所引文件无未提交改动 · HEAD cdd8125e09a2；均不自动代表线上版本。

                return {"state": "installed_cached", "runtime_acceptance": "not_run"}

文件 SHA-256：65e47b064bdf780fc14cdb2841370356c633a0b9343fb47e90be1a768f791c80

用户数据、原生HOME与凭证路径独立于程序目录；不声明该工作副本等于线上字节。

发布仓工作副本 · 所引文件无未提交改动 · HEAD cdd8125e09a2；均不自动代表线上版本。

export MEYO_LEGION_DIR="$DATA"

文件 SHA-256：799d7e17a3990b592b8f798d50be7ae28190277566de954557a80ad1e2c49808

使用SQLite备份/完整性/恢复后逻辑摘要与文件核对；只证明脚本能力，不执行线上回滚。

发布仓工作副本 · 所引文件无未提交改动 · HEAD cdd8125e09a2；均不自动代表线上版本。

    "live_restore_performed": False,

文件 SHA-256：3a0c6e42f85174272bcdf1d582a398661377bf828d0ecb510845ec6518e4b937