feat(owner-attention): notify owners of terminal Todo failures

docs/product/periodic-reports.md

@@ -422,6 +422,25 @@ account acceptance.
 | --- | --- |
 | ![Partial notification coverage in a narrow viewport](../assets/personal-workspace/owner-attention-source/partial-phone.png) | ![Recovered notification worker in a narrow viewport](../assets/personal-workspace/owner-attention-source/recovered-phone.png) |
 
+### Terminal-failure notices
+
+A blocked Todo that is neither done nor superseded and has a `reject` or `cancel`
+in its typed `decision_scope_outcomes` becomes a separate owner-attention subject,
+`<goal>/<todo>/terminal`, independent of its ordinary owner-action subject.
+The notice uses the latest terminal outcome by recorded time when all terminal
+outcomes have timestamps; otherwise it uses the last terminal outcome in the list.
+Its digest includes the terminal outcome count, so a second outcome starts a new
+notice generation even when the latest terms are unchanged. Clearing the terminal
+outcomes, leaving blocked status, completing or superseding the Todo withdraws
+unsent notices; delivered receipts remain unchanged. A missing `decision_scope`
+does not suppress a terminal notice. These notices reuse **Automatically notify
+new owner tasks**, its first-enable baseline and the existing outbound worker;
+they add no setting, scheduler or message consumer.
+
+未完成、未被替代的阻塞待办若有 `reject` 或 `cancel` 结果，会成为独立的所有者关注项；缺少 `decision_scope` 也不会漏报。
+终止结果均有时间戳时取最新一项，否则取列表最后一项；计数变化会产生新一代通知。结果清除或待办离开阻塞、完成、被替代时撤回未发送通知，已送达回执不变。
+此通知复用“自动通知新增所有者任务”、首次启用时的历史基线和现有出站工作器，不新增设置、调度器或消息消费者。
+
 The intended flow is:
 
 ```text

loopx/control_plane/presentation/owner_attention.ts

@@ -136,6 +136,26 @@ export function planOwnerAttention(input: JsonObject): JsonObject {
       if (current.has(key)) throw new Error("duplicate owner Todo");
       current.set(key, {key, goal_id: goalId, todo_id: todoId, terms, digest: digest(terms), active: true});
     }
+    // Failure rows are complete active blocked-terminal snapshots from Core.
+    for (const failure of records(goal.failures ?? [], "failures")) {
+      const todoId = requireNonEmptyString(failure.todo_id, "failure.todo_id");
+      if (!["reject", "cancel"].includes(String(failure.outcome))) throw new Error("invalid terminal outcome");
+      const count = requireInteger(failure.terminal_outcome_count ?? 1, "terminal_outcome_count");
+      if (count < 1) throw new Error("invalid terminal outcome count");
+      const terms: JsonObject = {text: requireNonEmptyString(failure.text, "failure.text"),
+        note: failure.outcome, reason: failure.reason ?? "", terminal_outcome_count: count};
+      if (failure.terminal_outcome_identities != null) {
+        // Canonical hashes ignore object key order; sorting ignores outcome list order.
+        terms.terminal_outcome_identities = records(failure.terminal_outcome_identities, "terminal_outcome_identities")
+          .map(row => Object.fromEntries(["decision_scope", "outcome", "source_todo_id"]
+            .filter(key => row[key] != null).map(key => [key, row[key]])))
+          .sort((a, b) => digest(a).localeCompare(digest(b)));
+      }
+      if (failure.evidence != null && failure.evidence !== "") terms.evidence = failure.evidence;
+      const key = `${goalId}/${todoId}/terminal`;
+      if (current.has(key)) throw new Error("duplicate terminal Todo");
+      current.set(key, {key, goal_id: goalId, todo_id: todoId, terms, digest: digest(terms), active: true});
+    }
   }
   const subjects: JsonObject[] = [...current.values()].map(row => {
     const old = previous.find(item => item.key === row.key);
@@ -145,10 +165,12 @@ export function planOwnerAttention(input: JsonObject): JsonObject {
       const id = digest([row.key, generation, row.digest]);
       const terms = row.terms as JsonObject;
       // Structured scopes remain identity/authority inputs, not operator copy.
-      const body = [`待你处理 · ${row.goal_id}`, String(terms.text),
+      const terminal = terms.terminal_outcome_count != null;
+      const body = [terminal ? `任务失败 · ${row.goal_id}` : `待你处理 · ${row.goal_id}`, String(terms.text),
         ...["note", "reason", "evidence"].flatMap(key =>
           terms[key] == null ? [] : [`${({note: "说明", reason: "原因", evidence: "依据"} as Record<string, string>)[key]}：${
             typeof terms[key] === "string" ? terms[key] : JSON.stringify(terms[key])}`]),
+        ...(terminal ? [`终止决定共 ${terms.terminal_outcome_count} 项；需要所有者决定。`] : []),
         "可在原管家对话中查看或答复这项待办。"].join("\n\n");
       outbox.push({id, key: row.key, generation, digest: row.digest, goal_id: row.goal_id,
         todo_id: row.todo_id, body, status: Buffer.byteLength(body, "utf8") > 20000 ? "failed" : "pending",

loopx/extensions/lark/owner_attention.py

@@ -10,6 +10,7 @@
 import json
 import re
 import time
+from datetime import datetime
 from pathlib import Path
 from typing import Any
 
@@ -140,7 +141,8 @@ def _snapshot(self, binding: dict[str, Any]) -> dict[str, Any]:
         goals, unavailable, failures = [], [], []
         for goal_id in scope:
             try:
-                goals.append({"goal_id": goal_id, "todos": self._goal_todos(goal_id)})
+                goals.append({"goal_id": goal_id, "todos": self._goal_todos(goal_id),
+                              "failures": self._goal_failures(goal_id)})
             except (ValueError, OSError, KeyError, TypeError, RuntimeError) as exc:
                 unavailable.append(goal_id)
                 # Only native error codes reach health; exception messages may
@@ -166,6 +168,45 @@ def _goal_todos(self, goal_id: str) -> list[dict[str, Any]]:
             todos.append(detail["todo"])
         return todos
 
+    def _goal_failures(self, goal_id: str) -> list[dict[str, Any]]:
+        controller = self.transport.core.controller
+        args = dict(registry_path=controller.registry_path, goal_id=goal_id,
+                    runtime_root_arg=str(controller.coordination_runtime_root), status="blocked")
+        listing = list_goal_todos(**args)
+        if listing.get("ok") is not True or listing.get("todo_count") != len(listing.get("todos", [])):
+            raise ValueError("the native terminal Todo source is incomplete")
+        failures = []
+        for reference in listing["todos"]:
+            # List rows may be compact. Exact reads own complete outcome terms.
+            detail = list_goal_todos(**args, todo_id=reference["todo_id"])
+            if detail.get("ok") is not True or not detail.get("matched") or not isinstance(detail.get("todo"), dict):
+                raise ValueError("a native terminal Todo changed while reading")
+            todo = detail["todo"]
+            if todo.get("status") != "blocked" or todo.get("done") or todo.get("superseded_by"):
+                continue
+            outcomes = [row for row in todo.get("decision_scope_outcomes", [])
+                        if row.get("outcome") in {"reject", "cancel"}]
+            if not outcomes:
+                continue
+            # Without complete timestamps the authoritative list order wins.
+            latest = outcomes[-1]
+            if all(row.get("recorded_at") for row in outcomes):
+                latest = max(enumerate(outcomes), key=lambda pair: (
+                    datetime.fromisoformat(pair[1]["recorded_at"].replace("Z", "+00:00")).timestamp(), pair[0]))[1]
+            failure = {"todo_id": todo["todo_id"], "text": todo["text"], "outcome": latest["outcome"],
+                       "reason": latest.get("reason", todo.get("reason", "")),
+                       "terminal_outcome_count": len(outcomes),
+                       # Only typed identity fields participate; never copy private outcome prose.
+                       "terminal_outcome_identities": sorted(
+                           [{key: row[key] for key in ("decision_scope", "outcome", "source_todo_id")
+                             if key in row} for row in outcomes],
+                           key=lambda row: json.dumps(row, sort_keys=True))}
+            evidence = latest.get("evidence", todo.get("evidence"))
+            if evidence is not None:
+                failure["evidence"] = evidence
+            failures.append(failure)
+        return failures
+
     @staticmethod
     def _plan(state: dict[str, Any], snapshot: dict[str, Any]) -> dict[str, Any]:
         return dict(effect_runtime_result("presentation.owner_attention.plan",

tests/control_plane_ts/goal_attention.test.ts

@@ -1,3 +1,4 @@
+import {planOwnerAttention} from "../../loopx/control_plane/presentation/owner_attention.ts";
 import assert from "node:assert/strict";
 import {test} from "node:test";
 import type {JsonObject} from "../../loopx/control_plane/effect_program.ts";
@@ -152,3 +153,19 @@ test("preview retains unknown source status and omitted work without claiming a
   assert.deepEqual(result.goals[0].attention, {...attention, details_omitted: true});
   assert.throws(() => boundGoalAttention({goals: [], preview: "true"}));
 });
+
+test("pending answers retain live owner work whose proactive notice is baseline suppressed or held", () => {
+  const empty = {schema_version: "lark_owner_attention_v0", revision: 0, enabled: true,
+    initialized: false, subjects: [], outbox: []};
+  const initial = planOwnerAttention({state: empty, source_complete: true,
+    goals: [{goal_id: "alpha", todos: [gate]}]});
+  assert.deepEqual(initial.outbox, [], "first enable suppresses historical notices");
+  const discovered = planOwnerAttention({state: initial, source_complete: true,
+    goals: [{goal_id: "alpha", todos: [gate, {...gate, todo_id: "todo_second"}]}]});
+  const held = planOwnerAttention({state: discovered, goals: [], scope_goal_ids: ["alpha"],
+    unavailable_goal_ids: ["alpha"], source_complete: false});
+  assert.equal((held.outbox as JsonObject[])[0].source_current, false);
+  const result = projectGoalAttention({goal_id: "alpha", todos: [gate, {...gate, todo_id: "todo_second"}], blockers: []});
+  assert.deepEqual(result.items.map(item => item.todo_id), ["todo_gate", "todo_second"]);
+  assert.ok(result.items.every(item => item.owner_must_act === true));
+});

tests/control_plane_ts/owner_attention.test.ts

@@ -1,3 +1,4 @@
+import {planUserCompletion} from "../../loopx/control_plane/todos/user_completion.ts";
 import assert from "node:assert/strict";
 import test from "node:test";
 import {observeOwnerAttention, ownerAttentionHealth, ownerAttentionReferences, planOwnerAttention} from "../../loopx/control_plane/presentation/owner_attention.ts";
@@ -234,3 +235,91 @@ test("source diagnostics identify the unavailable Goal and retain no raw error d
   assert.deepEqual(current.source_failures, []);
   assert.equal(rows(current).length, 1, "old-writer metadata cannot retire or duplicate the original notice");
 });
+
+const terminal = {todo_id: "todo_failed", text: "Build the preview", outcome: "reject",
+  reason: "Review rejected the proposal", evidence: "https://example.org/review", terminal_outcome_count: 1};
+const failurePlan = (state: JsonObject, failures: JsonObject[]) => planOwnerAttention({state,
+  goals: [{goal_id: "alpha", todos: [], failures}], source_complete: true});
+
+test("reject and cancel each create one stable terminal subject with complete terms", () => {
+  for (const outcome of ["reject", "cancel"]) {
+    const failure = {...terminal, outcome};
+    const state = failurePlan(failurePlan(empty(), []), [failure]);
+    assert.equal(rows(state).length, 1);
+    const notice = rows(state)[0];
+    assert.equal(notice.key, "alpha/todo_failed/terminal");
+    const {text, reason, evidence, terminal_outcome_count} = failure;
+    assert.equal(notice.digest, canonicalAuthoritySha256({text, note: outcome, reason, evidence, terminal_outcome_count}));
+    for (const term of [text, outcome, reason, evidence, "需要所有者决定", "1 项"]) assert.ok(String(notice.body).includes(term));
+    assert.deepEqual(failurePlan(JSON.parse(JSON.stringify(state)), [failure]), state);
+    const changed = failurePlan(state, [{...failure, reason: "Revised review reason"}]);
+    assert.equal(rows(changed).length, 2);
+    assert.equal(rows(changed)[0].status, "superseded");
+    assert.equal(rows(changed)[1].generation, 2);
+  }
+});
+
+test("a second terminal outcome changes generation even when latest terms are otherwise identical", () => {
+  const first = failurePlan(failurePlan(empty(), []), [terminal]);
+  const second = failurePlan(first, [{...terminal, terminal_outcome_count: 2}]);
+  assert.equal(rows(second).length, 2);
+  assert.equal(rows(second)[0].status, "superseded");
+  assert.equal(rows(second)[1].generation, 2);
+  assert.notEqual(rows(second)[0].digest, rows(second)[1].digest);
+  assert.ok(String(rows(second)[1].body).includes("2 项"));
+});
+
+test("resolved, deferred, superseded or cleared terminal outcomes withdraw unsent failure notices", () => {
+  for (const change of [{status: "done"}, {status: "deferred"}, {status: "blocked", superseded_by: "todo_new"},
+    {status: "blocked", decision_scope_outcomes: []}]) {
+    const pending = failurePlan(failurePlan(empty(), []), [terminal]);
+    const resolved = planOwnerAttention({state: pending, source_complete: true,
+      goals: [{goal_id: "alpha", todos: [{...task, todo_id: terminal.todo_id, role: "agent", ...change}], failures: []}]});
+    assert.equal(rows(resolved)[0].status, "superseded");
+    assert.equal((resolved.subjects as JsonObject[])[0].active, false);
+  }
+});
+
+test("two failures remain distinct and terminal size failures retain the existing cap", () => {
+  const base = failurePlan(empty(), []);
+  const state = failurePlan(base, [terminal, {...terminal, todo_id: "todo_second", outcome: "cancel"}]);
+  assert.deepEqual(rows(state).map(row => row.key), ["alpha/todo_failed/terminal", "alpha/todo_second/terminal"]);
+  const long = rows(failurePlan(base, [{...terminal, reason: "题".repeat(9000)}]))[0];
+  assert.equal(long.status, "failed");
+  assert.equal(long.error_code, "notice_content_too_large");
+});
+
+test("absent failures preserves the ordinary plan byte for byte", () => {
+  const base = plan(empty(), []);
+  for (const todos of [[], [reviewTask], [{...task, status: "done"}]]) {
+    const legacy = plan(base, todos);
+    const explicit = planOwnerAttention({state: base, source_complete: true,
+      goals: [{goal_id: "alpha", todos, failures: []}]});
+    assert.equal(JSON.stringify(legacy), JSON.stringify(explicit));
+    assert.ok(!JSON.stringify(legacy).includes("/terminal"));
+  }
+});
+
+
+test("native in-place terminal scope replacement advances generation and identical identities stay stable", () => {
+  const scope = {schema_version: "decision_scope_v0", kind: "direction", granularity: "action", scope_key: "preview"};
+  const gate = {todo_id: "todo_review_a", role: "user", task_class: "user_gate", status: "open",
+    unblocks_todo_id: terminal.todo_id, decision_scope: scope};
+  let target: JsonObject = {todo_id: terminal.todo_id, role: "agent", status: "blocked"};
+  target = {...target, ...planUserCompletion(gate, [target], "reject").updates};
+  target = {...target, ...planUserCompletion({...gate, todo_id: "todo_review_b",
+    decision_scope: {...scope, scope_key: "publish"}}, [target], "reject").updates};
+  const project = (row: JsonObject): JsonObject => ({...terminal, terminal_outcome_count: 2,
+    terminal_outcome_identities: row.decision_scope_outcomes});
+  const first = failurePlan(failurePlan(empty(), []), [project(target)]);
+  assert.equal(rows(first)[0].generation, 1);
+  target = {...target, ...planUserCompletion(gate, [target], "cancel").updates};
+  assert.deepEqual((target.decision_scope_outcomes as JsonObject[]).map(row => row.outcome), ["cancel", "reject"]);
+  const second = failurePlan(first, [project(target)]);
+  assert.equal(rows(second)[1].generation, 2);
+  assert.notEqual(rows(second)[0].digest, rows(second)[1].digest);
+  assert.equal(rows(second)[0].body, rows(second)[1].body);
+  assert.deepEqual(failurePlan(second, [project(target)]), second);
+  const reordered = {...project(target), terminal_outcome_identities: [...target.decision_scope_outcomes as JsonObject[]].reverse()};
+  assert.deepEqual(failurePlan(second, [reordered]), second);
+});