fix(topic-memory): stop head conflicts from spending window attempts

docs/en/development/processing-failures.md

@@ -83,7 +83,7 @@ Topic Memory operation-level coverage:
 | `topic-memory` stages | Output parsing | Partial | Provider schema failures use the shared mapping; semantic rejections are business validation. |
 | `topic-memory` stages | Business validation | Covered | Every `TopicMemoryGenerationError` code maps to bounded diagnostics: domain and model rejections carry a stage and, when evidence supports it, a reason; control-flow rejections carry only `failure_location`. Input-budget failures carry input counts, token estimates, estimator identity, and the effective budget; the window token ceiling carries the charged tokens and the ceiling, while attempt and request ceilings are reported through their block details. |
 | `topic-memory.index` and query embedding | Embedding | Covered | A missing embedding adapter and a wrong vector count are classified. |
-| Publish / persistence | Persistence | Partial | Projection content and vector-completeness rules are business validation; storage-invariant failures map to `persistence` / `storage_failure`; unsupported capabilities report `configuration`; publisher conflicts keep their domain outcomes, and other storage failures keep unknown. |
+| Publish / persistence | Persistence | Partial | Projection content and vector-completeness rules are business validation; storage-invariant failures map to `persistence` / `storage_failure`; unsupported capabilities report `configuration`; publisher conflicts keep their domain outcomes, and other storage failures keep unknown. A Topic head conflict records `head_conflict` as the attempt's last error and does not spend a window attempt. |
 | Usage accounting | Usage accounting | Missing | The window provider budget is reported as `generation` / `usage_limit`; accounting failures are not classified. |
 
 Experience, Profile, Skill, and Handoff operation-level coverage:

docs/en/docs/operate/configuration.md

@@ -302,7 +302,10 @@ Topic Workers enforce a durable allowance per unadvanced Scope Cursor: 3 attempt
 and 64,000,000 estimated token-capacity units across all retries. A Window admits at most 4,194,304 canonical evidence
 characters including metadata; nested input is also bounded. Exhaustion preserves Sources, Cursor, Pending, and the
 same-Scope tail, and stops further provider calls. Flush and restart do not reset it; inspect
-`pc_topic_memory_work_budgets` and structured errors for operator remediation.
+`pc_topic_memory_work_budgets` and structured errors for operator remediation. A head conflict, where a concurrent
+replace or Dream moves a Topic head after the Worker read it, does not spend an attempt: the Worker records
+`head_conflict` as that attempt's last error, rereads the head and recomputes. Its reserved requests and token capacity
+stay charged, so a head that keeps moving stops at `window_provider_budget_exceeded` instead of retrying indefinitely.
 
 Topic generation accepts `max_tokens`, `temperature`, `top_p`, `top_k`, `seed`, `presence_penalty`, `frequency_penalty`,
 `timeout`, `openai_reasoning_effort`, `openai_text_verbosity`, `service_tier`, `openai_service_tier`,

docs/en/docs/operate/observability.md

@@ -128,6 +128,8 @@ the supervisor records interruption or failure instead.
 
 The supervisor emits `artifact_processing.queued` on local queue admission, `artifact_processing.dispatched`
 when it starts an attempt, and `artifact_processing.attempt.completed` when it finalizes that attempt.
+An attempt that ends in a control conflict finalizes with outcome `conflict`; its `error_code` is `cursor_conflict` or
+`head_conflict`.
 `queue_wait_ms` uses a monotonic clock and excludes earlier policy waits and retry backoff before queue admission.
 On completion, `duration_ms` measures the supervisor attempt separately from queue time and child execution time.
 These are execution diagnostics, not counts of newly published Artifacts.
@@ -488,8 +490,8 @@ The dashboard labels a policy block as “被规则阻止” (“blocked by a ru
 another failure; it means the same Source frontier has reached its retry or resource allowance and the system stopped
 starting more model work. Common `error_code` values are:
 
-- `window_attempt_limit`: the same frontier has used three attempts and has not completed. Request and token budgets
-  may still be available.
+- `window_attempt_limit`: the same frontier has used three attempts and has not completed. Attempts that lost a
+  Topic head conflict to a concurrent replace or Dream are not counted. Request and token budgets may still be available.
 - `window_provider_budget_exceeded`: the same frontier reached its request or token budget.
 - `source_complexity_limit`: the Source window exceeds the processing limit.
 

docs/en/docs/workflows/topic-memory.md

@@ -103,6 +103,13 @@ that no longer owns the current frontier returns HTTP 409. Cursor, target and So
 and its retry timestamp remain readable after progress resumes. Normal flushes and restarts retain their existing
 budget behavior. Each explicitly accepted retry grants the same three-attempt and provider ceilings.
 
+A manual replace or Dream can move a Topic head while a Worker is generating against it. The Worker's publication then
+fails its head check and writes nothing; the Worker records `head_conflict` as the attempt's last error, and the next
+attempt rereads the head and recomputes. Such an attempt does not count toward the three-attempt ceiling, but its
+provider reservations do. A head that keeps moving therefore ends with `window_provider_budget_exceeded`. That
+reason is usually also the block's `last_error_code`, because the ceiling is crossed during an attempt; a low
+`attempts` count shows that recomputations after head conflicts, not failed attempts, used the budget.
+
 The equivalent Python Client SDK methods are `get_topic_memory_processing_status`,
 `get_topic_memory_processing_block`, and `retry_topic_memory_processing`. Their HTTP routes are
 `POST /v1/topic-memory/processing/status`, `POST /v1/topic-memory/processing/blocks/get`, and

docs/zh/development/processing-failures.md

@@ -77,7 +77,7 @@ Topic Memory 的 operation 级覆盖：
 | `topic-memory` 各阶段 | Output parsing | Partial | provider schema 失败走共享映射；语义拒绝归为 business validation。 |
 | `topic-memory` 各阶段 | Business validation | Covered | 每个 `TopicMemoryGenerationError` code 都有有界诊断：领域/模型拒绝携带 stage，并在有证据时给出 reason；控制流拒绝只携带 `failure_location`。输入预算失败携带输入计数、token 估算、estimator 标识和生效预算；window token 上限触发时携带已计费 token 与上限，attempt/request 上限通过 block details 报告。 |
 | `topic-memory.index` 与查询嵌入 | Embedding | Covered | 缺少 embedding adapter、向量数量错误均已分类。 |
-| Publish / persistence | Persistence | Partial | 投影内容及向量完整性规则归业务校验；存储不变量失败归 `persistence` / `storage_failure`；不支持的能力归 `configuration`；publisher 冲突保留既有领域结果，其他存储失败保持 unknown。 |
+| Publish / persistence | Persistence | Partial | 投影内容及向量完整性规则归业务校验；存储不变量失败归 `persistence` / `storage_failure`；不支持的能力归 `configuration`；publisher 冲突保留既有领域结果，其他存储失败保持 unknown。Topic head 冲突将 `head_conflict` 记为该次尝试的最后错误，不消耗 window 尝试次数。 |
 | Usage accounting | Usage accounting | Missing | window provider budget 归为 `generation` / `usage_limit`；账务失败未分类。 |
 
 Experience、Profile、Skill 与 Handoff 的 operation 级覆盖：

docs/zh/docs/operate/configuration.md

@@ -273,7 +273,10 @@ flush 超时后统计可能尚不完整，在一个 Server 上读取统计也不
 Topic Worker 对尚未推进的 Scope Cursor 强制使用持久额度：跨全部重试最多 3 次尝试、512 次预留 provider 请求和
 64,000,000 个估算 token 容量单位。Window 的 canonical evidence（包含 metadata）最多 4,194,304 个字符，并限制
 嵌套复杂度。耗尽后保留 Source、Cursor、Pending 和同 Scope 尾部，停止后续 provider 调用；flush 和重启均不重置。
-运维可检查 `pc_topic_memory_work_budgets` 与结构化错误以明确修复。
+运维可检查 `pc_topic_memory_work_budgets` 与结构化错误以明确修复。Worker 读取 Topic head 后，如果并发的 replace
+或 Dream 移动了该 head（head 冲突），这次尝试不计入尝试次数：Worker 将 `head_conflict` 记为该次尝试的最后错误，
+重新读取 head 后重新计算。该次尝试预留的请求数和 token 容量仍然计入，因此 head 持续变化时会在
+`window_provider_budget_exceeded` 处停止，不会无限重试。
 
 Topic generation 只允许有界标量设置：`max_tokens`、`temperature`、`top_p`、`top_k`、`seed`、`presence_penalty`、
 `frequency_penalty`、`timeout`、`openai_reasoning_effort`、`openai_text_verbosity`、`service_tier`、

docs/zh/docs/operate/observability.md

@@ -123,6 +123,7 @@ Child 返回结果不能单独证明业务完成。强制终止进程可能来
 
 Supervisor 在本地入队时输出 `artifact_processing.queued`，开始一次执行时输出
 `artifact_processing.dispatched`，结束该次执行时输出 `artifact_processing.attempt.completed`。
+以控制冲突结束的执行，其 outcome 为 `conflict`，`error_code` 为 `cursor_conflict` 或 `head_conflict`。
 `queue_wait_ms` 使用单调时钟，不包含入队前的策略等待和重试退避。
 完成事件的 `duration_ms` 单独表示 supervisor 的执行耗时，与排队时间和 child 执行时间分开计算。
 这些事件用于执行诊断，不能当作新发布 Artifact 的数量。
@@ -434,7 +435,8 @@ Scope 的后台处理结果尚未确认写入成功；长时间大于零可能
 “策略阻塞”在页面中显示为“被规则阻止”。它不是 Worker 崩溃，也不是一次新的失败；它表示同一批 Source
 已经达到允许的重试或资源预算，系统停止继续调用模型。常见 `error_code` 如下：
 
-- `window_attempt_limit`：同一 Cursor 后的这批 Source 已尝试 3 次，仍未完成。请求数和 Token 预算不一定用完。
+- `window_attempt_limit`：同一 Cursor 后的这批 Source 已尝试 3 次，仍未完成。因并发 replace 或 Dream 导致
+  Topic head 冲突的尝试不计入。请求数和 Token 预算不一定用完。
 - `window_provider_budget_exceeded`：同一批 Source 已达到请求数或 Token 预算。
 - `source_complexity_limit`：该窗口的 Source 结构超过处理上限。
 

docs/zh/docs/workflows/topic-memory.md

@@ -147,6 +147,12 @@ retry 需要 `scope.admin` 权限，查询还需要该角色未包含的 `scope.
 恢复后仍可读取原阻塞记录和重试时间。普通 flush、重启的额度语义不变；每次明确接受的 retry 使用相同的三次尝试
 和 provider 上限。
 
+Worker 生成期间，手动 replace 或 Dream 可能移动它所基于的 Topic head。此时 Worker 的发布无法通过 head 检查，
+不写入任何内容；Worker 将 `head_conflict` 记为该次尝试的最后错误，下一次尝试重新读取 head 后重新计算。这类尝试
+不计入三次尝试上限，但其 provider 预留仍然计入。因此 head 持续变化时会以 `window_provider_budget_exceeded`
+结束。由于额度通常在某次尝试中途耗尽，阻塞记录的 `last_error_code` 一般也是该原因；`attempts` 很小说明预算主要
+消耗在 head 冲突后的重新计算上，而不是失败的尝试。
+
 Python Client SDK 对应方法为 `get_topic_memory_processing_status`、`get_topic_memory_processing_block` 和
 `retry_topic_memory_processing`。HTTP 路径分别为 `POST /v1/topic-memory/processing/status`、
 `POST /v1/topic-memory/processing/blocks/get` 和 `POST /v1/topic-memory/processing/retry`。这些管理操作不作为 MCP 工具提供。

src/powercontext/builtin/persistence/topic_memory_budget.py

@@ -63,6 +63,20 @@
 })
 
 
+# The last error of an attempt whose publication lost the Topic head CAS to a concurrent
+# replace or Dream. The next attempt rereads the head and recomputes (RFC 1614 section 5.3).
+TOPIC_MEMORY_HEAD_CONFLICT = "head_conflict"
+
+
+def _spent_attempts(row: Mapping[Any, Any]) -> int:
+    """Attempts charged to the ceiling; a head conflict hands its attempt to the recomputation.
+
+    The conflicted attempt's provider reservations stay charged, so the request and token
+    ceilings still bound how often a moving head can make the window recompute.
+    """
+    return int(row["attempts"]) - (row.get("last_error_code") == TOPIC_MEMORY_HEAD_CONFLICT)
+
+
 def exhausted_reason(row: Mapping[Any, Any]) -> str:
     """Why this frontier stops, or "" while another attempt may still be spent.
 
@@ -73,7 +87,7 @@ def exhausted_reason(row: Mapping[Any, Any]) -> str:
     code = str(row["failure_code"])
     if code and code not in RECOVERABLE_TOPIC_MEMORY_REJECTIONS:
         return code
-    if row["attempts"] >= MAX_TOPIC_MEMORY_WORK_ATTEMPTS:
+    if _spent_attempts(row) >= MAX_TOPIC_MEMORY_WORK_ATTEMPTS:
         # Name the rejection that kept recurring; the ceiling only says how many
         # attempts were spent on it.
         return code or "window_attempt_limit"
@@ -276,7 +290,7 @@ async def begin(self) -> None:
                     update(table)
                     .where(*self._key())
                     .values(
-                        attempts=row["attempts"] + 1,
+                        attempts=_spent_attempts(row) + 1,
                         attempt_id=self.attempt_id,
                         source_through=max(self.source_through, row["source_through"]),
                     )

src/powercontext/builtin/runtime/artifact_processing.py

@@ -1580,6 +1580,7 @@ async def _reap(self, state: _FamilyState) -> None:  # noqa: C901 - distinct Wor
                     ArtifactProcessingWorkerOutcome.HEAD_CONFLICT,
                 }:
                     observed_outcome = "conflict"
+                    diagnostic_code = completion.outcome.value
                     previous = state.retries.get(scope)
                     self._defer(
                         state,

src/powercontext/builtin/runtime/topic_memory_processing.py

@@ -105,6 +105,7 @@
     MAX_TOPIC_MEMORY_WORK_REQUESTS,
     MAX_TOPIC_MEMORY_WORK_TOKENS,
     RECOVERABLE_TOPIC_MEMORY_REJECTIONS,
+    TOPIC_MEMORY_HEAD_CONFLICT,
     TopicMemoryWorkBudget,
     require_topic_memory_work_available,
 )
@@ -572,14 +573,22 @@ async def process(
                         await self._generate(assignment.scope_id, projected) if projected else ((), {})
                     )
                     operations = await self._prepare_operations(assignment.scope_id, proposals, candidates, evidence)
-                with (
-                    nullcontext()
-                    if self._tracing is None
-                    else self._tracing.stage("topic_memory.commit", attributes={})
-                ):
-                    await self._publisher.publish(
-                        assignment, evidence, operations, work_budget=budget, commit_hook=commit_hook
-                    )
+                try:
+                    with (
+                        nullcontext()
+                        if self._tracing is None
+                        else self._tracing.stage("topic_memory.commit", attributes={})
+                    ):
+                        await self._publisher.publish(
+                            assignment, evidence, operations, work_budget=budget, commit_hook=commit_hook
+                        )
+                except RevisionConflictError:
+                    # A concurrent replace or Dream moved a Topic head after this attempt read it.
+                    # Recording the cause hands the attempt back; its reservations stay charged. Only a
+                    # generated proposal can target an existing head, and every stage reserves before its
+                    # call, so each recomputation spends from the frontier's request and token ceilings.
+                    await budget.record_failure(TOPIC_MEMORY_HEAD_CONFLICT)
+                    return ArtifactProcessingWorkerCompletion(ArtifactProcessingWorkerOutcome.HEAD_CONFLICT)
             except TopicMemoryGenerationError as error:
                 if error.code in RECORDED_TOPIC_MEMORY_REJECTIONS:
                     await budget.fail(error.code)
@@ -591,6 +600,7 @@ async def process(
         except GenerationConflictError:
             return ArtifactProcessingWorkerCompletion(ArtifactProcessingWorkerOutcome.CURSOR_CONFLICT)
         except RevisionConflictError:
+            # Not raised by publication, so this attempt keeps counting toward the ceiling.
             return ArtifactProcessingWorkerCompletion(ArtifactProcessingWorkerOutcome.HEAD_CONFLICT)
         finally:
             self._document_vectors.reset(vector_token)