refactor: cut edge-case complexity and dedupe protocol/agent layers

Executes the optimization plan in CODE_REVIEW.md (sections 3-11): net
~-2,000 lines against the pre-plan working tree while keeping every
security baseline (three-state ACL, labeled tokens, label-transition
fact retraction, shared LLM budget) intact.

- Includes the previously uncommitted 2026-07-11 P1/P2 fixes that
  survived the plan: tail-anchored recall-meta split (P1-1),
  label-transition retract_digested (P1-2), "formation" collection
  alias (P1-3), shared conversation-read guard (P2-1), shared LLM
  semaphore across HTTP+MCP (P2-2).
- Reverts edge-case defenses per the accepted-risk register
  (CODE_REVIEW.md section 9): update snapshot-rollback, dual-sided
  formation/maintenance recheck, per-version poison-fuse map, indented
  code-block chunk rules, OKF drift patching / zero-growth replay /
  resource deletion propagation, eval harness_degradation +
  OptimizeAbort + transport degradations, CBOR error bodies,
  verify-failure audit events, digest retraction backfill, lazy census,
  and more.
- Structural dedup: new src/authz.rs as the single HTTP+MCP
  authorization source (4 AuthzMode); shared RunnerHost /
  drive_runner_loop skeleton for formation/maintenance; MCP mirror
  types replaced by JsonSchema derives (schema structurally identical);
  eval spaces moved to a forked InMemory store; llm_semaphore into
  AppState; shared bearer/shard header parsing (fixes HTTP/MCP drift).
- API.md/API_cn.md: error bodies are always JSON; OKF can no longer
  delete resource (use wiki_commit).

Tests: 254 lib + 21 bin green; fmt/clippy clean. Deleted tests belong
to the reverted defenses; security-baseline assertions unchanged.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: zensh

anda_brain/API.md

@@ -8,6 +8,7 @@
 - Supported serialization formats:
   - Request: `Content-Type: application/json | application/cbor | text/markdown`
   - Response: `Accept: application/json | application/cbor | text/markdown`
+  - Content negotiation applies to success bodies only; error response bodies are always JSON regardless of `Accept`
 - Most business endpoints return an RPC envelope: `RpcResponse<T>`
 - MCP clients can use the built-in Streamable HTTP endpoint: `/mcp/<space_id>`, or the local stdio server: `anda_brain mcp --space-id <space_id> [local|aws]`
 
@@ -602,27 +603,27 @@ When `ED25519_PUBKEYS` is set, configure the remote MCP client with an `Authoriz
 ### GET `/v1/{space_id}/conversations/{conversation_id}?collection=<collection>`
 
 - Purpose: Get a single conversation detail
-- Auth: SpaceToken/CWT `read` (public spaces are unauthenticated; private spaces require a valid token); tokens restricted by ACL labels are rejected with `403` — conversations persist the full agent runner history, which is not label-scoped
+- Auth: SpaceToken/CWT `read` (public spaces are unauthenticated; private spaces require a valid token); tokens restricted by ACL labels are rejected with `403` — conversations persist the full agent runner history, which is not label-scoped; `collection=recall` additionally rejects anonymous access on public spaces with `403` (recall runs from a private era may embed labeled wiki content)
 - Query:
-  - `collection?: string` // "recall" or "maintenance" for non-default conversation collections; unknown values are rejected with `400`
+  - `collection?: string` // "formation" (default), "recall" or "maintenance"; unknown values are rejected with `400`
 - Response: `RpcResponse<Conversation>`
 
 ### GET `/v1/{space_id}/conversations/{conversation_id}/delta?collection=<collection>&messages_offset=<n>&artifacts_offset=<n>`
 
 - Purpose: Get incremental conversation updates after client-side offsets
-- Auth: SpaceToken/CWT `read` (public spaces are unauthenticated; private spaces require a valid token); tokens restricted by ACL labels are rejected with `403`
+- Auth: SpaceToken/CWT `read` (public spaces are unauthenticated; private spaces require a valid token); tokens restricted by ACL labels are rejected with `403` (`collection=recall`: anonymous access on public spaces is also rejected)
 - Query:
-  - `collection?: string` // "recall" or "maintenance" for non-default conversation collections; unknown values are rejected with `400`
+  - `collection?: string` // "formation" (default), "recall" or "maintenance"; unknown values are rejected with `400`
   - `messages_offset?: number` // returns only messages after this offset, defaults to `0`
   - `artifacts_offset?: number` // returns only artifacts after this offset, defaults to `0`
 - Response: `RpcResponse<ConversationDelta>`
 
 ### GET `/v1/{space_id}/conversations?collection=<collection>&cursor=<cursor>&limit=<n>`
 
 - Purpose: List conversations with pagination
-- Auth: SpaceToken/CWT `read` (public spaces are unauthenticated; private spaces require a valid token); tokens restricted by ACL labels are rejected with `403`
+- Auth: SpaceToken/CWT `read` (public spaces are unauthenticated; private spaces require a valid token); tokens restricted by ACL labels are rejected with `403` (`collection=recall`: anonymous access on public spaces is also rejected)
 - Query:
-  - `collection?: string` // "recall" or "maintenance" for non-default conversation collections; unknown values are rejected with `400`
+  - `collection?: string` // "formation" (default), "recall" or "maintenance"; unknown values are rejected with `400`
   - `cursor?: string`
   - `limit?: number`
 - Response: `RpcResponse<Conversation[]>` (next page cursor is returned via `next_cursor`)
@@ -827,3 +828,4 @@ if (recall.error) {
 - Authentication failure: HTTP `401`, response body is `RpcError`
 - Invalid request/parameters: HTTP `400`, response body is `RpcError`
 - Success: HTTP `200`, response body is usually `RpcResponse<T>`
+- Error response bodies are always JSON, even when the request asked for `application/cbor` or `text/markdown` (this includes the wiki `409` conflict body carrying `error.data.current_version`); only success bodies follow the `Accept` header

anda_brain/API_cn.md

@@ -8,6 +8,7 @@
 - 支持的序列化格式：
   - 请求：`Content-Type: application/json | application/cbor | text/markdown`
   - 响应：`Accept: application/json | application/cbor | text/markdown`
+  - 内容协商仅作用于成功响应体；错误响应体始终为 JSON，与 `Accept` 无关
 - 大多数业务接口都会返回 RPC 包装后的结构体：`RpcResponse<T>`
 - MCP 客户端可使用内置的支持流式传输的 HTTP MCP 端点：`/mcp/<space_id>`，也可以使用本地 stdio server：`anda_brain mcp --space-id <space_id> [local|aws]`
 
@@ -602,27 +603,27 @@ MCP_AUTH_TOKEN="$SPACE_TOKEN" \
 ### GET `/v1/{space_id}/conversations/{conversation_id}?collection=<collection>`
 
 - 作用：获取单条会话详情
-- 鉴权：SpaceToken/CWT `read`（公开空间免鉴权，私有空间需有效 token）；带 ACL 标签限制的 token 返回 `403`——会话持久化了完整的 agent 运行历史，不受标签过滤
+- 鉴权：SpaceToken/CWT `read`（公开空间免鉴权，私有空间需有效 token）；带 ACL 标签限制的 token 返回 `403`——会话持久化了完整的 agent 运行历史，不受标签过滤；`collection=recall` 对公开空间的匿名访问也返回 `403`（私有期的 recall 运行可能内嵌 labeled wiki 内容）
 - Query:
-  - `collection?: string` // "recall" 或 "maintenance" 指定非默认会话集合；未知值返回 `400`
+  - `collection?: string` // "formation"（默认）、"recall" 或 "maintenance"；未知值返回 `400`
 - 响应：`RpcResponse<Conversation>`
 
 ### GET `/v1/{space_id}/conversations/{conversation_id}/delta?collection=<collection>&messages_offset=<n>&artifacts_offset=<n>`
 
 - 作用：按客户端已消费的 offset 获取会话增量更新
-- 鉴权：SpaceToken/CWT `read`（公开空间免鉴权，私有空间需有效 token）；带 ACL 标签限制的 token 返回 `403`
+- 鉴权：SpaceToken/CWT `read`（公开空间免鉴权，私有空间需有效 token）；带 ACL 标签限制的 token 返回 `403`（`collection=recall`：公开空间匿名访问同样拒绝）
 - Query:
-  - `collection?: string` // "recall" 或 "maintenance" 指定非默认会话集合；未知值返回 `400`
+  - `collection?: string` // "formation"（默认）、"recall" 或 "maintenance"；未知值返回 `400`
   - `messages_offset?: number` // 仅返回该偏移量之后的新消息，默认 `0`
   - `artifacts_offset?: number` // 仅返回该偏移量之后的新 artifacts，默认 `0`
 - 响应：`RpcResponse<ConversationDelta>`
 
 ### GET `/v1/{space_id}/conversations?collection=<collection>&cursor=<cursor>&limit=<n>`
 
 - 作用：分页列出会话
-- 鉴权：SpaceToken/CWT `read`（公开空间免鉴权，私有空间需有效 token）；带 ACL 标签限制的 token 返回 `403`
+- 鉴权：SpaceToken/CWT `read`（公开空间免鉴权，私有空间需有效 token）；带 ACL 标签限制的 token 返回 `403`（`collection=recall`：公开空间匿名访问同样拒绝）
 - Query:
-  - `collection?: string` // "recall" 或 "maintenance" 指定非默认会话集合；未知值返回 `400`
+  - `collection?: string` // "formation"（默认）、"recall" 或 "maintenance"；未知值返回 `400`
   - `cursor?: string`
   - `limit?: number`
 - 响应：`RpcResponse<Conversation[]>`（并通过 `next_cursor` 给出下一页游标）
@@ -830,3 +831,4 @@ if (recall.error) {
 - 认证失败：HTTP `401`，响应体为 `RpcError`
 - 参数错误：HTTP `400`，响应体为 `RpcError`
 - 成功时：HTTP `200`，响应体通常为 `RpcResponse<T>`
+- 错误响应体始终为 JSON，即使请求指定了 `application/cbor` 或 `text/markdown`（包括携带 `error.data.current_version` 的 wiki `409` 冲突响应体）；只有成功响应体遵循 `Accept` 协商

anda_brain/src/agents.rs

@@ -8,6 +8,7 @@ use anda_db::schema::DocumentId;
 use anda_engine::{
     context::CompletionRunner,
     memory::{Conversation, ConversationStatus},
+    unix_ms,
 };
 use parking_lot::RwLock;
 use std::collections::VecDeque;
@@ -42,6 +43,199 @@ const COMPACTION_CONTINUE_PROMPT: &str = "Continue the active memory-agent work
 /// persisted conversation.
 pub(super) const PERSIST_EVERY_N_TURNS: usize = 5;
 
+/// Hard guardrails for the formation/maintenance runner loops (review P2-3).
+///
+/// A model that keeps emitting tool calls without converging would otherwise
+/// run forever: the runner compacts every ~81 turns and simply continues, so
+/// the agent's processing flag stays set and the space's memory system stalls
+/// (formation's queue freezes; maintenance blocks formation entirely). Both
+/// agents legitimately need far more turns than recall
+/// (`RECALL_MAX_MODEL_TURNS` = 7 + 180s), so these caps are intentionally
+/// loose. Exceeding either budget marks the conversation Failed through the
+/// host's `mark_failed`, which reuses each agent's existing failure path
+/// (formation: the Failed retry path; maintenance: releases the processing
+/// slot through the guard/hook flow). Budgets are checked between turns so an
+/// in-flight KIP write turn is never cancelled halfway.
+pub(super) const RUNNER_MAX_MODEL_TURNS: usize = 200;
+pub(super) const RUNNER_MAX_WALL_CLOCK_MS: u64 = 30 * 60 * 1000;
+
+/// Control flow returned by [`RunnerHost::after_turn`].
+pub(super) enum RunnerFlow {
+    /// Keep looping (the host may have queued a follow-up).
+    Continue,
+    /// The conversation converged; leave the runner loop cleanly.
+    Break,
+}
+
+/// Per-agent seams of the shared formation/maintenance runner loop
+/// ([`drive_runner_loop`]). Everything else — budget guards, the compaction
+/// arm, turn accounting, `append_runner_history`, the status three-state, the
+/// persistence throttle, the single failure exit with usage backfill and the
+/// tail Working flush — is identical between the two agents and lives in the
+/// driver. recall deliberately does not use this skeleton: its timeout
+/// wrapper and failed_output semantics do not fit these seams.
+pub(super) trait RunnerHost {
+    /// Agent label used in budget-exceeded failure reasons.
+    fn label(&self) -> &'static str;
+
+    /// The agent's completed-conversation ring served as model context.
+    fn history(&self) -> &RwLock<VecDeque<Document>>;
+
+    /// Persists the current full conversation snapshot; failures are logged
+    /// by the host and must not interrupt the processing loop.
+    async fn persist_snapshot(&self, conversation: &Conversation);
+
+    /// Marks the conversation Failed with `reason` and persists it.
+    async fn mark_failed(&self, conversation: &mut Conversation, reason: String);
+
+    /// Host-specific convergence predicate for the current turn (formation
+    /// also treats an idle runner as done once the review pass has run).
+    fn turn_is_done(&self, runner: &CompletionRunner) -> bool;
+
+    /// Runs on every successful (non-failed) turn, before the completed
+    /// history push (formation clears `failed_reason` so the Failed-retry
+    /// path converges to a clean snapshot; maintenance leaves it untouched).
+    fn on_turn_success(&self, conversation: &mut Conversation);
+
+    /// Runs at the end of every non-terminal turn, after persistence. The
+    /// host decides whether the loop ends now (formation injects the pending
+    /// review follow-up when the runner idles and otherwise breaks as soon as
+    /// it is done) or defers to the runner's own `Ok(None)` exit on the next
+    /// iteration (maintenance).
+    fn after_turn(&mut self, runner: &mut CompletionRunner, is_done: bool) -> RunnerFlow;
+}
+
+/// Shared runner loop driving one formation/maintenance conversation to a
+/// terminal state. Every failure exits through the labeled break so usage
+/// backfill and `mark_failed` live at exactly one place below the loop.
+pub(super) async fn drive_runner_loop<H: RunnerHost>(
+    host: &mut H,
+    runner: &mut CompletionRunner,
+    conversation: &mut Conversation,
+) {
+    let started_at_ms = unix_ms();
+    let mut replace_initial_input = true;
+    let mut persisted_runner_history_len = 0;
+    let mut total_model_turns = 0usize;
+    let mut accounted_runner_turns = 0usize;
+    let mut unpersisted_turns = 0usize;
+    let failure: Option<String> = 'run: {
+        loop {
+            // Guardrails against a non-converging tool loop; see
+            // RUNNER_MAX_MODEL_TURNS. Exceeding a budget takes the host's
+            // existing mark_failed path.
+            if total_model_turns >= RUNNER_MAX_MODEL_TURNS {
+                break 'run Some(format!(
+                    "{} exceeded model turn limit of {}",
+                    host.label(),
+                    RUNNER_MAX_MODEL_TURNS
+                ));
+            }
+            if unix_ms().saturating_sub(started_at_ms) >= RUNNER_MAX_WALL_CLOCK_MS {
+                break 'run Some(format!(
+                    "{} exceeded wall-clock budget of {} seconds",
+                    host.label(),
+                    RUNNER_MAX_WALL_CLOCK_MS / 1000
+                ));
+            }
+
+            match compact_runner_if_needed(runner).await {
+                Ok(true) => {
+                    // The compaction handoff consumed one model turn, and the
+                    // replacement runner restarts its own turn counter.
+                    total_model_turns = total_model_turns.saturating_add(1);
+                    accounted_runner_turns = runner.turns();
+                    persisted_runner_history_len = 0;
+                    replace_initial_input = false;
+                }
+                Ok(false) => {}
+                Err(err) => break 'run Some(format!("CompletionRunner error: {err:?}")),
+            }
+
+            match runner.next().await {
+                Ok(None) => break 'run None,
+                Ok(Some(res)) => {
+                    let runner_turns = runner.turns();
+                    total_model_turns = total_model_turns
+                        .saturating_add(runner_turns.saturating_sub(accounted_runner_turns));
+                    accounted_runner_turns = runner_turns;
+
+                    let now_ms = unix_ms();
+                    let is_done = host.turn_is_done(runner);
+
+                    append_runner_history(
+                        conversation,
+                        &res.chat_history,
+                        &mut persisted_runner_history_len,
+                        &mut replace_initial_input,
+                    );
+
+                    conversation.status = if res.failed_reason.is_some() {
+                        ConversationStatus::Failed
+                    } else if is_done {
+                        ConversationStatus::Completed
+                    } else {
+                        ConversationStatus::Working
+                    };
+                    conversation.usage = res.usage;
+                    conversation.updated_at = now_ms;
+
+                    if let Some(failed_reason) = res.failed_reason {
+                        conversation.failed_reason = Some(failed_reason);
+                    } else {
+                        host.on_turn_success(conversation);
+                        push_completed_history(host.history(), conversation, 2);
+                    }
+
+                    // Persisting rewrites the full message array (O(turns^2)
+                    // over a session), so intermediate Working turns are
+                    // throttled; terminal statuses always persist. See
+                    // PERSIST_EVERY_N_TURNS.
+                    unpersisted_turns = unpersisted_turns.saturating_add(1);
+                    if conversation.status != ConversationStatus::Working
+                        || unpersisted_turns >= PERSIST_EVERY_N_TURNS
+                    {
+                        host.persist_snapshot(conversation).await;
+                        unpersisted_turns = 0;
+                    }
+
+                    if conversation.status == ConversationStatus::Cancelled
+                        || conversation.status == ConversationStatus::Failed
+                    {
+                        break 'run None;
+                    }
+
+                    if let RunnerFlow::Break = host.after_turn(runner, is_done) {
+                        break 'run None;
+                    }
+                }
+                Err(err) => break 'run Some(format!("CompletionRunner error: {err:?}")),
+            }
+        }
+    };
+
+    // Single failure exit. The usage snapshot happens after the error
+    // occurred: failure can strike after usage was accumulated but before it
+    // was copied from a runner output (e.g. a compaction handoff consumed ~a
+    // full context window of input tokens and the very next call errors), so
+    // the runner's running total is backfilled here, like recall does, or
+    // those tokens vanish from the agent's usage ledger. Success exits must
+    // NOT backfill: the runner's final_output/final_idle_output mem::take
+    // total_usage, so the last `res.usage` already carries the full total.
+    if let Some(reason) = failure {
+        conversation.usage = runner.total_usage().clone();
+        host.mark_failed(conversation, reason).await;
+    }
+
+    // Terminal and failure exits above always persist (any non-Working
+    // status forces a write, and mark_failed writes its own snapshot), so
+    // only a Working exit — e.g. the runner returning `Ok(None)` — can still
+    // hold turns skipped by the throttle.
+    if unpersisted_turns > 0 && conversation.status == ConversationStatus::Working {
+        host.persist_snapshot(conversation).await;
+    }
+}
+
 fn queued_runner_tokens(runner: &CompletionRunner) -> u64 {
     runner
         .steering_message_iter()
@@ -52,21 +246,15 @@ fn queued_runner_tokens(runner: &CompletionRunner) -> u64 {
 
 pub(super) async fn compact_runner_if_needed(
     runner: &mut CompletionRunner,
-    extra_pending_tokens: u64,
-    continue_after_compaction: bool,
 ) -> Result<bool, BoxError> {
-    if !runner
-        .needs_compaction_with(|| queued_runner_tokens(runner).saturating_add(extra_pending_tokens))
-    {
+    if !runner.needs_compaction_with(|| queued_runner_tokens(runner)) {
         return Ok(false);
     }
 
     let (mut compacted, output) = runner.handoff(None).await?;
     compacted.accumulate(&output.usage);
     compacted.accumulate_tools_usage(&output.tools_usage);
-    if continue_after_compaction {
-        compacted.follow_up(ContentPart::from(COMPACTION_CONTINUE_PROMPT.to_string()));
-    }
+    compacted.follow_up(ContentPart::from(COMPACTION_CONTINUE_PROMPT.to_string()));
     *runner = compacted;
     Ok(true)
 }