From 6c7744464e3b936d55d0e12e0f22519d7fd21753 Mon Sep 17 00:00:00 2001 From: LUOSENGWA Date: Sat, 29 Aug 2026 15:26:20 +0000 Subject: [PATCH 1/3] feat(teamharness): auto-send Matrix completion notification on submit_task The taskflow MCP's delegate_task already publishes the assignment to the task room at the code layer (_send_delegate_notification, stable txn), but submit_task only returns a notificationNeeded hint - the Worker must self-remember to @mention the leader with the contract line. Real deployments (Node1) lost this line after multi-turn sessions, leaving the Leader with no resume signal. Mirror the delegate pattern in the same file: - _send_task_completion_notification: Matrix HTTP PUT with m.mentions (same path/auth as the message tool), stable txn 'submit-' so a retry cannot duplicate, first line follows the task-execution skill contract (TASK_COMPLETED/BLOCKED + result path). - _task_completion_notification: best-effort orchestration - resolve the leader from the runtime config team roster, reuse the recorded completionEventId on retry, validate room membership, persist the event id on success. A send failure returns {sent: false, error} and never blocks the terminal submission. - submit_task response gains 'notification'; the notificationNeeded hint is kept (it also drives the requester reply-route report). Contract tests (test-taskflow.rb): completion line + mentions + worker + summary + event-id persistence, retry reuses without duplicating, BLOCKED status uses the BLOCKED line, forced 500 does not block the submission and persists no event id. Context file-event selection made mxcUri-based instead of positional. --- plugins/teamharness/mcp/server.py | 176 +++++++++++++++ .../teamharness/mcp/tools/test-taskflow.rb | 206 +++++++++++++++++- 2 files changed, 378 insertions(+), 4 deletions(-) diff --git a/plugins/teamharness/mcp/server.py b/plugins/teamharness/mcp/server.py index 4ab6b15b8..a2a03eda7 100644 --- a/plugins/teamharness/mcp/server.py +++ b/plugins/teamharness/mcp/server.py @@ -4079,6 +4079,174 @@ def _send_delegate_notification( return {"sent": False, "error": f"Matrix API error: {exc}"} +def _team_leader_matrix_id() -> str: + """Resolve the team leader's Matrix user ID from the runtime config. + + The controller projects the full team roster (with roles and Matrix + user IDs) into the worker's runtime config. Returns an empty string + when no leader entry exists (standalone runs) so callers can skip + the notification instead of failing. + """ + config = _load_runtime_config() + team = _section(config, "team") + members = team.get("members") + if not isinstance(members, list): + return "" + for member in members: + if not isinstance(member, dict): + continue + role = str(member.get("role") or "").strip().lower().replace("_", "-") + if role in {"team-leader", "teamleader", "leader"}: + return str(member.get("matrixUserId") or member.get("matrix_user_id") or "").strip() + return "" + + +def _send_task_completion_notification( + arguments: dict[str, Any], + *, + room_id: str, + task_id: str, + status: str, + summary: str, + leader: str, + worker: str = "", + result_path: str = "", +) -> dict[str, Any]: + """Send the automatic Worker completion notification for submit_task. + + Publishes the completion line to the Task room with ``m.mentions`` + using the same Matrix HTTP send path as the message tool. The first + line follows the task-execution skill contract so leader-side prompts + that parse completion lines keep working: + @leader TASK_COMPLETED: - Result: shared/tasks//result.md + @leader BLOCKED: - + The transaction ID is stable per task so a retry cannot produce a + duplicate completion. + """ + homeserver = os.getenv("AGENTTEAMS_MATRIX_URL", "").rstrip("/") + token = os.getenv("AGENTTEAMS_WORKER_MATRIX_TOKEN", "") + if not homeserver or not token: + return { + "sent": False, + "error": "AGENTTEAMS_MATRIX_URL and AGENTTEAMS_WORKER_MATRIX_TOKEN are required", + } + + matrix_room_id = str(room_id or "").strip() + if matrix_room_id.startswith("room:"): + matrix_room_id = matrix_room_id[len("room:") :].strip() + if not matrix_room_id.startswith("!"): + return {"sent": False, "error": f"invalid Matrix room target: {room_id}"} + + summary_preview = (summary or "")[:500] + if len(summary or "") > 500: + summary_preview += "..." + if status == "BLOCKED": + notification_text = f"{leader} BLOCKED: {task_id} - {summary_preview}" + detail = "" + else: + if result_path: + notification_text = f"{leader} TASK_COMPLETED: {task_id} - Result: {result_path}" + else: + notification_text = f"{leader} TASK_COMPLETED: {task_id} - {summary_preview}" + detail = f"\n{summary_preview}" if summary_preview else "" + if worker: + notification_text += f"\n- Worker: {worker}" + if status in {"REVISION_NEEDED", "INTERRUPTED"}: + notification_text += f"\n- Status: {status}" + notification_text += detail + mentions = [leader] + content = _matrix_content(notification_text, mentions) + + room_enc = urllib.parse.quote(matrix_room_id, safe="") + txn = urllib.parse.quote(f"submit-{task_id}", safe="") + url = f"{homeserver}/_matrix/client/v3/rooms/{room_enc}/send/m.room.message/{txn}" + request = urllib.request.Request( + url, + data=json.dumps(content).encode("utf-8"), + headers={ + "Authorization": f"Bearer {token}", + "Content-Type": "application/json", + }, + method="PUT", + ) + try: + with urllib.request.urlopen(request, timeout=10) as response: + data = json.loads(response.read().decode("utf-8") or "{}") + event_id = str(data.get("event_id") or "").strip() + if not event_id: + return {"sent": False, "error": "Matrix send returned no event_id"} + return { + "sent": True, + "eventId": event_id, + "roomId": matrix_room_id, + "leader": leader, + } + except urllib.error.HTTPError as exc: + body = exc.read().decode("utf-8", errors="replace")[:200] + return { + "sent": False, + "error": f"Matrix API error: HTTP {exc.code}: {body}", + } + except (urllib.error.URLError, TimeoutError, OSError) as exc: + return {"sent": False, "error": f"Matrix API error: {exc}"} + + +def _task_completion_notification( + arguments: dict[str, Any], + task: dict[str, Any], + task_id: str, + status: str, + summary: str, +) -> dict[str, Any]: + """Best-effort completion notification orchestration for submit_task. + + Mirrors the delegate_task notification lifecycle: resolve the leader + from the runtime config, reuse an already-recorded event on retry, + validate room membership, send, then persist the event id so the + notification cannot be duplicated. Any problem returns a skipped + result and never blocks the terminal submission. + """ + room_id = str(task.get("room_id") or "").strip() + if not room_id: + return {"sent": False, "skipped": True, "error": "task has no room_id"} + leader = _team_leader_matrix_id() + if not leader: + return { + "sent": False, + "skipped": True, + "error": "team leader Matrix ID not found in runtime config", + } + if task.get("completionEventId"): + return { + "sent": True, + "eventId": str(task["completionEventId"]), + "roomId": _canonical_room_id(room_id), + "leader": leader, + "reused": True, + } + membership = _validate_assignee_membership(room_id, leader) + if not membership.get("ok"): + return { + "sent": False, + "skipped": True, + "error": str(membership.get("error") or "room membership check failed"), + } + notification = _send_task_completion_notification( + arguments, + room_id=room_id, + task_id=task_id, + status=status, + summary=summary, + leader=leader, + worker=str(task.get("assigned_to") or ""), + result_path=str(task.get("result_path") or ""), + ) + if notification.get("sent"): + task["completionEventId"] = notification.get("eventId") + _write_task(arguments, task) + return notification + + def _taskflow(arguments: dict[str, Any]) -> dict[str, Any]: action = str(arguments.get("action") or "").strip() payload = _payload(arguments) @@ -4394,6 +4562,13 @@ def _taskflow(arguments: dict[str, Any]) -> dict[str, Any]: deliverables, _attachment_parent_event_id(payload, arguments), ) + notification = _task_completion_notification( + arguments, + task, + task_id, + status, + summary, + ) return { "ok": True, "tool": "taskflow", @@ -4401,6 +4576,7 @@ def _taskflow(arguments: dict[str, Any]) -> dict[str, Any]: "task": task, "publishedArtifacts": published_artifacts, "synced": _sync_task(arguments, task_id, exclude=["spec.md", "base/"]), + "notification": notification, "notificationNeeded": _notification_needed( "submit_task", {"project_id": task.get("project_id", "")}, diff --git a/plugins/tests/teamharness/mcp/tools/test-taskflow.rb b/plugins/tests/teamharness/mcp/tools/test-taskflow.rb index 0479e76dc..09d50f0b9 100644 --- a/plugins/tests/teamharness/mcp/tools/test-taskflow.rb +++ b/plugins/tests/teamharness/mcp/tools/test-taskflow.rb @@ -78,7 +78,18 @@ def fail!(message) } runtime_config = pathlib.Path("#{root}") / "runtime.yaml" runtime_config.write_text( - "team:\\n teamRoomId: '!team:example.test'\\n", + "team:\\n" + " teamRoomId: '!team:example.test'\\n" + " leaderRuntimeName: 'admin'\\n" + " members:\\n" + " - name: 'Admin'\\n" + " runtimeName: 'admin'\\n" + " role: 'team_leader'\\n" + " matrixUserId: '@admin:example.test'\\n" + " - name: 'Worker A'\\n" + " runtimeName: 'worker-a'\\n" + " role: 'worker'\\n" + " matrixUserId: '@worker-a:example.test'\\n", encoding="utf-8", ) os.environ["TEAMHARNESS_RUNTIME_CONFIG"] = str(runtime_config) @@ -142,6 +153,11 @@ def do_PUT(self): self.end_headers() self.wfile.write(json.dumps({"errcode": "M_UNKNOWN", "error": "forced Matrix failure"}).encode("utf-8")) return + if "/send/m.room.message/submit-" in parsed.path and os.environ.get("TEAMHARNESS_TEST_FAIL_SUBMIT_NOTIFICATION") == "1": + self.send_response(500) + self.end_headers() + self.wfile.write(json.dumps({"errcode": "M_UNKNOWN", "error": "forced Matrix failure"}).encode("utf-8")) + return matrix["events"].append({ "path": parsed.path, "auth": self.headers.get("Authorization"), @@ -410,6 +426,56 @@ def block_yaml_import(name, *args, **kwargs): if not all(upload.get("auth") == "Bearer test-token" for upload in matrix["uploads"][:2]): raise AssertionError(f"Matrix upload auth mismatch: {matrix['uploads']!r}") + def completion_events(): + return [ + event for event in matrix["events"] + if "/send/m.room.message/submit-t-001" in event["path"] + ] + + first_completion = completion_events() + if len(first_completion) != 1: + raise AssertionError(f"submit_task should send exactly one completion notification: {matrix['events']!r}") + completion_body = first_completion[0]["content"].get("body", "") + if "@admin:example.test" not in (first_completion[0]["content"].get("m.mentions") or {}).get("user_ids", []): + raise AssertionError(f"completion notification must mention the leader: {first_completion[0]['content']!r}") + if "TASK_COMPLETED: t-001 - Result: shared/tasks/t-001/result.md" not in completion_body: + raise AssertionError(f"completion notification must carry the contract line: {completion_body!r}") + if "- Worker: @worker-a:example.test" not in completion_body: + raise AssertionError(f"completion notification must carry the executor: {completion_body!r}") + if "Input collected." not in completion_body: + raise AssertionError(f"completion notification must carry the summary: {completion_body!r}") + if first_completion[0]["auth"] != "Bearer test-token": + raise AssertionError(f"completion notification auth mismatch: {first_completion[0]['auth']!r}") + submitted_meta = json.loads((pathlib.Path("#{workspace}") / f"shared/tasks/{task_id}/meta.json").read_text(encoding="utf-8")) + if not submitted_meta.get("completionEventId"): + raise AssertionError(f"submit_task did not persist completionEventId: {submitted_meta!r}") + if submitted_meta.get("completionEventId") != submitted.get("notification", {}).get("eventId"): + raise AssertionError(f"persisted completionEventId mismatch: {submitted_meta!r} vs {submitted.get('notification')!r}") + + resubmitted = payload("taskflow", { + "role": "worker", + "action": "submit_task", + "payload": { + "taskId": task_id, + "status": "SUCCESS", + "summary": "Input collected.", + "parentEventId": "$task-parent", + "deliverables": [ + "shared/tasks/t-001/result.md", + "shared/tasks/t-001/workspace/analysis.md", + ], + }, + }) + if not resubmitted.get("ok") or resubmitted["task"]["status"] != "submitted": + raise AssertionError(f"resubmit_task failed: {resubmitted!r}") + resubmit_notification = resubmitted.get("notification") or {} + if resubmit_notification.get("reused") is not True: + raise AssertionError(f"resubmit should reuse the recorded completion notification: {resubmit_notification!r}") + if resubmit_notification.get("eventId") != submitted.get("notification", {}).get("eventId"): + raise AssertionError(f"resubmit notification event id mismatch: {resubmit_notification!r}") + if len(completion_events()) != 1: + raise AssertionError(f"resubmit must not duplicate the completion notification: {completion_events()!r}") + context_project_id = "context-parent-project" context_task_id = "context-parent-task" payload("projectflow", { @@ -476,9 +542,19 @@ def block_yaml_import(name, *args, **kwargs): raise AssertionError(f"context submit_task should publish result artifact: {context_submitted!r}") if context_published[0].get("parentEventId") != "$context-task-parent": raise AssertionError(f"context submit_task did not infer parent event: {context_submitted!r}") - context_event = matrix["events"][-1]["content"] - if context_event.get("m.relates_to") != {"rel_type": "com.agentteams.attachment", "event_id": "$context-task-parent"}: - raise AssertionError(f"context submit_task file event missing attachment relation: {context_event!r}") + context_file_event = next( + ( + event["content"] + for event in reversed(matrix["events"]) + if event["content"].get("msgtype") == "m.file" + and event["content"].get("url") == context_published[0].get("mxcUri") + ), + None, + ) + if context_file_event is None: + raise AssertionError(f"context submit_task file event not found: {context_published!r}") + if context_file_event.get("m.relates_to") != {"rel_type": "com.agentteams.attachment", "event_id": "$context-task-parent"}: + raise AssertionError(f"context submit_task file event missing attachment relation: {context_file_event!r}") secret_task_id = "secret-artifact-01" payload("taskflow", { @@ -524,6 +600,128 @@ def block_yaml_import(name, *args, **kwargs): if any("abcdefghijklmnopqrstuvwxyz1234567890" in upload.get("body", "") for upload in matrix["uploads"]): raise AssertionError("sensitive value leaked into Matrix upload") + # --- Completion notification: BLOCKED status carries the BLOCKED contract line. --- + blocked_project_id = "blocked-project" + blocked_task_id = "blocked-task" + payload("projectflow", { + "action": "create_project", + "payload": { + "projectId": blocked_project_id, + "title": "Blocked Project", + "replyRoute": { + "channel": "matrix", + "targetUser": "@admin:example.test", + "targetSession": "!team:example.test", + }, + }, + }) + payload("projectflow", { + "action": "plan_dag", + "payload": { + "projectId": blocked_project_id, + "tasks": [{ + "taskId": blocked_task_id, + "title": "Blocked task", + "assignedTo": "@worker-a:example.test", + "dependsOn": [], + }], + }, + }) + payload("taskflow", { + "role": "leader", + "action": "delegate_task", + "payload": { + "projectId": blocked_project_id, + "taskId": blocked_task_id, + "roomId": "room:!team:example.test", + "spec": "Will be blocked.", + }, + }) + blocked_submitted = payload("taskflow", { + "role": "worker", + "action": "submit_task", + "payload": { + "taskId": blocked_task_id, + "status": "BLOCKED", + "summary": "GPU OOM on node 2, needs 24G context.", + }, + }) + if not blocked_submitted.get("ok") or blocked_submitted["task"]["status"] != "submitted": + raise AssertionError(f"blocked submit_task failed: {blocked_submitted!r}") + blocked_events = [ + event for event in matrix["events"] + if "/send/m.room.message/submit-blocked-task" in event["path"] + ] + if len(blocked_events) != 1: + raise AssertionError(f"blocked submit should send one completion notification: {matrix['events']!r}") + blocked_body = blocked_events[0]["content"].get("body", "") + if "BLOCKED: blocked-task - GPU OOM on node 2, needs 24G context." not in blocked_body: + raise AssertionError(f"blocked notification must carry the BLOCKED contract line: {blocked_body!r}") + if "TASK_COMPLETED" in blocked_body: + raise AssertionError(f"blocked notification must not claim completion: {blocked_body!r}") + + # --- Failure injection: a completion send failure must NOT block the + # terminal submission (best-effort by contract). --- + fail_submit_project_id = "fail-submit-project" + fail_submit_task_id = "fail-submit-task" + payload("projectflow", { + "action": "create_project", + "payload": { + "projectId": fail_submit_project_id, + "title": "Fail Submit Project", + "replyRoute": { + "channel": "matrix", + "targetUser": "@admin:example.test", + "targetSession": "!team:example.test", + }, + }, + }) + payload("projectflow", { + "action": "plan_dag", + "payload": { + "projectId": fail_submit_project_id, + "tasks": [{ + "taskId": fail_submit_task_id, + "title": "Fail submit task", + "assignedTo": "@worker-a:example.test", + "dependsOn": [], + }], + }, + }) + payload("taskflow", { + "role": "leader", + "action": "delegate_task", + "payload": { + "projectId": fail_submit_project_id, + "taskId": fail_submit_task_id, + "roomId": "room:!team:example.test", + "spec": "Matrix is down during submit.", + }, + }) + os.environ["TEAMHARNESS_TEST_FAIL_SUBMIT_NOTIFICATION"] = "1" + try: + fail_submit_result = payload("taskflow", { + "role": "worker", + "action": "submit_task", + "payload": { + "taskId": fail_submit_task_id, + "status": "SUCCESS", + "summary": "Result ready but Matrix is down.", + }, + }) + finally: + os.environ.pop("TEAMHARNESS_TEST_FAIL_SUBMIT_NOTIFICATION", None) + if not fail_submit_result.get("ok") or fail_submit_result["task"]["status"] != "submitted": + raise AssertionError(f"completion notification failure must not block submission: {fail_submit_result!r}") + fail_submit_notification = fail_submit_result.get("notification") or {} + if fail_submit_notification.get("sent") is not False: + raise AssertionError(f"failed completion send must report sent=False: {fail_submit_notification!r}") + if "HTTP 500" not in str(fail_submit_notification.get("error", "")): + raise AssertionError(f"failed completion send must report the Matrix error: {fail_submit_notification!r}") + fail_submit_meta = json.loads((pathlib.Path("#{workspace}") / f"shared/tasks/{fail_submit_task_id}/meta.json").read_text(encoding="utf-8")) + if fail_submit_meta.get("completionEventId"): + raise AssertionError(f"failed completion send must not persist completionEventId: {fail_submit_meta!r}") + # --- Failure injection: a notification send failure must return a # retryable error and must NOT leave the task assigned. --- fail_project_id = "fail-inject-project" From 1f0676bf00df4dd9957df57ab8102ee2bd4558ab Mon Sep 17 00:00:00 2001 From: LUOSENGWA Date: Sat, 29 Aug 2026 15:33:25 +0000 Subject: [PATCH 2/3] docs(design): task completion notification design doc (MCP layer) --- docs/design/task-completion-notification.md | 132 ++++++++++++++++++++ 1 file changed, 132 insertions(+) create mode 100644 docs/design/task-completion-notification.md diff --git a/docs/design/task-completion-notification.md b/docs/design/task-completion-notification.md new file mode 100644 index 000000000..83ee9e305 --- /dev/null +++ b/docs/design/task-completion-notification.md @@ -0,0 +1,132 @@ +# Task Completion Notification (submit_task) + +> **Status**: Implemented in `plugins/teamharness/mcp/server.py` (branch +> `fix/task-completion-notification`). +> **Scope**: The TeamHarness MCP `taskflow` tool — the task lifecycle +> executed by Worker/Leader runtimes. The Manager-runtime hook variant +> (`copaw_worker.hooks.tools.taskflow`) is out of scope: the Manager +> coordinates tasks but is not a task executor, so its `submit_task` +> path is not the production completion route. + +## Problem + +The taskflow MCP layer is asymmetric: + +- `delegate_task` **atomically** records the assignment in task state and + publishes the assignment to the task room with `m.mentions` + (`_send_delegate_notification`, stable txn `delegate-`, + recorded `eventId`, reuse-on-retry). +- `submit_task` records the terminal state and auto-publishes result + artifacts as `m.file` events (`_publish_task_artifacts`), but the + completion *message* is only a `notificationNeeded` **hint** — the + Worker's LLM must self-remember to send + `@leader TASK_COMPLETED: - Result: shared/tasks//result.md`. + +Real deployments (Node1, multi-turn sessions) show the Worker forgetting +that line after context compaction. Consequences: + +- The Leader receives no wake signal: `check_task` is poll-based, and + `m.file` artifact events carry no mention, so the Leader is not + triggered by the artifacts alone. +- Downstream tasks stay stuck in `waiting` until the Leader happens to + poll or a human pokes the room. + +## Design + +Mirror the existing delegate pattern, same file, same send path: + +| Concern | `delegate_task` (existing) | `submit_task` (this change) | +|:--|:--|:--| +| Send helper | `_send_delegate_notification` | `_send_task_completion_notification` | +| Matrix path | HTTP PUT `/rooms/{room}/send/m.room.message/{txn}` (same as message tool) | identical | +| Credentials | `AGENTTEAMS_MATRIX_URL` + `AGENTTEAMS_WORKER_MATRIX_TOKEN` | identical (Worker's own token — the sender *is* the Worker) | +| Stable txn | `delegate-` | `submit-` | +| Mention | assignee mxid | **leader** mxid (resolved from runtime config) | +| Recorded event | task state `eventId` | task state `completionEventId` | +| Retry | reuse recorded `eventId` | reuse recorded `completionEventId` | +| Failure | task never marked `assigned` (hard fail) | **best-effort**: submission proceeds, response reports `sent: false` + error | + +Message contract (first line parseable by leader-side prompts; mirrors +the task-execution skill): + +``` +@leader TASK_COMPLETED: - Result: shared/tasks//result.md +- Worker: @worker:matrix.local + +``` + +``` +@leader BLOCKED: - +- Worker: @worker:matrix.local +``` + +`REVISION_NEEDED` / `INTERRUPTED` add a `- Status:` line; `SUCCESS` / +`SUCCESS_WITH_NOTES` / `BLOCKED` do not (the token already says it). + +### Leader resolution + +`_team_leader_matrix_id()` reads the runtime config the controller +projects into the Worker: `team.members[]` with +`role ∈ {team_leader, teamleader, leader}` → `matrixUserId` (same role +normalization as `_roomflow_room_meta`). Empty result → notification +skipped with a `skipped` reason (standalone runs). + +### Membership guard + +`_validate_assignee_membership(room_id, leader)` is reused as-is: when +the Matrix env is configured, the leader must be a joined member of the +task room; otherwise the send is skipped (reason recorded) instead of +producing an error event for a user who cannot receive it. + +### Idempotency + +- Stable txn `submit-`: Matrix de-duplicates a redelivered + identical PUT. +- `completionEventId` persisted in task state after the first success: + a later resubmit returns the recorded event with `reused: true` and + performs no HTTP call. + +### Best-effort by contract + +The completion message is a *notification*, not part of the state +mutation. Any problem (no room, no leader, no Matrix env, membership +missing, HTTP error) returns +`{"sent": false, "skipped"?: true, "error": "..."}` and the submit +still completes: `ok: true`, `status: "submitted"`, artifacts +published, state synced. The existing `notificationNeeded` hint is +**kept** — it also drives the requester reply-route report, which the +code-level line intentionally does not cover (different room/audience). + +## Changes + +| File | Change | +|:--|:--| +| `plugins/teamharness/mcp/server.py` | `_team_leader_matrix_id()`, `_send_task_completion_notification()`, `_task_completion_notification()`; `submit_task` branch calls the orchestrator after `_publish_task_artifacts` and adds `notification` to the response | +| `plugins/tests/teamharness/mcp/tools/test-taskflow.rb` | runtime config gains the team roster; fake Matrix server gains a `submit-` fault-injection branch; new assertions (see below); context file-event selection made mxcUri-based instead of positional (the last event is no longer guaranteed to be a file event) | + +## Tests (contract, `test-taskflow.rb`) + +1. **Send + content**: exactly one `submit-t-001` message event; + `m.mentions.user_ids` contains the leader; body carries the contract + line, `- Worker:` line, and the summary; auth = Worker token. +2. **Persistence**: task state `completionEventId` equals the response + `notification.eventId`. +3. **Retry**: resubmit with the same payload returns + `notification.reused: true` with the same `eventId` and sends no + second event. +4. **BLOCKED**: `BLOCKED: - ` line, no + `TASK_COMPLETED` text. +5. **Failure**: forced HTTP 500 on the `submit-` txn → submit still + `ok: true` / `submitted`, `notification.sent: false` with the HTTP + error, no `completionEventId` persisted. + +## Open questions + +1. **Should `check_task`'s polling be removed from leader prompts?** The + auto-notification makes blind polling redundant; keep it as a + reconciliation path (cheap) until the notification is proven in + production. +2. **Manager-runtime taskflow**: the same asymmetry exists in + `copaw_worker.hooks.tools.taskflow` (its `delegate_task` notifies, its + `submit_task` does not). Not addressed here — the Manager is not a + task executor; revisit if a deployment ever makes it one. From 6e36b41993571c4d185887d3d1d9f93c71da6aa8 Mon Sep 17 00:00:00 2001 From: LUOSENGWA Date: Sat, 5 Sep 2026 18:41:35 +0000 Subject: [PATCH 3/3] teamharness: lifecycle attention events (#1229 v2) - submit_task: per-status first-line tokens (TASK_COMPLETED / TASK_PARTIAL / TASK_REVISION_NEEDED / TASK_BLOCKED / TASK_FAILED), status validation, status-scoped txn and completionEventStatus so a changed-status resubmit sends a fresh event - @initiator routing: human members (task initiator) mentioned alongside the leader in completion / attention / project events - P0 ordering: sync shared storage before the completion notification; a failed sync withholds the event and returns a retryable failure - new taskflow action request_attention (worker/leader/remote-member): in-flight human decisions as first-class idempotent room events (kind approval/decision/escalation/other), terminal guard, sync-first, resolved by accept_task_result or explicit resolved=true - complete_project: code-level PROJECT_COMPLETED room event with idempotent projectCompletionEventId - task-execution SKILL.md: contract rewritten around the code-generated per-status events; full accepted status set - test-taskflow.rb: 12 new assertion groups (ordering, tokens, @initiator, validation, resubmit, request_attention, PROJECT_COMPLETED); mc shim gains a push-only sync-failure hook Local verification: full extracted harness green (all pre-existing + new assertions); pytest 73 passed (4 pre-existing env failures on base). Refs: https://github.com/agentscope-ai/AgentTeams/issues/1229 --- docs/design/task-completion-notification.md | 180 ++++++- plugins/teamharness/mcp/server.py | 507 ++++++++++++++++-- .../skills/team/task-execution/SKILL.md | 33 +- .../teamharness/mcp/tools/test-taskflow.rb | 312 +++++++++++ 4 files changed, 966 insertions(+), 66 deletions(-) diff --git a/docs/design/task-completion-notification.md b/docs/design/task-completion-notification.md index 83ee9e305..e89f6805e 100644 --- a/docs/design/task-completion-notification.md +++ b/docs/design/task-completion-notification.md @@ -1,4 +1,4 @@ -# Task Completion Notification (submit_task) +# Task Completion Notification (submit_task) + Lifecycle Attention Events > **Status**: Implemented in `plugins/teamharness/mcp/server.py` (branch > `fix/task-completion-notification`). @@ -7,6 +7,13 @@ > (`copaw_worker.hooks.tools.taskflow`) is out of scope: the Manager > coordinates tasks but is not a task executor, so its `submit_task` > path is not the production completion route. +> +> **v2 (PR review 2026-09-05, design issue #1229)** extends the original +> completion line into a full lifecycle attention model: +> per-status first-line tokens, `@initiator` (human members) routing, +> P0 sync-before-notify ordering with a retryable failure, a new +> `request_attention` action for in-flight human decisions, and a +> code-level `PROJECT_COMPLETED` event on `complete_project`. ## Problem @@ -40,14 +47,15 @@ Mirror the existing delegate pattern, same file, same send path: | Send helper | `_send_delegate_notification` | `_send_task_completion_notification` | | Matrix path | HTTP PUT `/rooms/{room}/send/m.room.message/{txn}` (same as message tool) | identical | | Credentials | `AGENTTEAMS_MATRIX_URL` + `AGENTTEAMS_WORKER_MATRIX_TOKEN` | identical (Worker's own token — the sender *is* the Worker) | -| Stable txn | `delegate-` | `submit-` | -| Mention | assignee mxid | **leader** mxid (resolved from runtime config) | +| Stable txn | `delegate-` | `submit--` (status-scoped: a same-status retry de-duplicates; a changed status produces a new event) | +| Mention | assignee mxid | **leader + human members** mxids (resolved from runtime config — the `@initiator` routing, see below) | | Recorded event | task state `eventId` | task state `completionEventId` | | Retry | reuse recorded `eventId` | reuse recorded `completionEventId` | | Failure | task never marked `assigned` (hard fail) | **best-effort**: submission proceeds, response reports `sent: false` + error | Message contract (first line parseable by leader-side prompts; mirrors -the task-execution skill): +the task-execution skill). Every accepted result status gets its own +first-line token so a leader prompt can branch on the line itself: ``` @leader TASK_COMPLETED: - Result: shared/tasks//result.md @@ -56,12 +64,34 @@ the task-execution skill): ``` ``` -@leader BLOCKED: - +@leader TASK_PARTIAL: - - Worker: @worker:matrix.local +- Status: PARTIAL ``` -`REVISION_NEEDED` / `INTERRUPTED` add a `- Status:` line; `SUCCESS` / -`SUCCESS_WITH_NOTES` / `BLOCKED` do not (the token already says it). +``` +@leader TASK_REVISION_NEEDED: - +- Worker: @worker:matrix.local +- Status: REVISION_NEEDED +``` + +``` +@leader TASK_BLOCKED: - +- Worker: @worker:matrix.local +- Status: BLOCKED +``` + +``` +@leader TASK_FAILED: - +- Worker: @worker:matrix.local +- Status: FAILED +``` + +`SUCCESS` / `SUCCESS_WITH_NOTES` keep the `TASK_COMPLETED` token and the +`Result:` line (no `- Status:` line — the token already says it); every +other token carries the `- Status:` line. `submit_task` validates the +submitted status against the accepted set and rejects unknown values +with a clear error instead of rendering a generic line. ### Leader resolution @@ -71,6 +101,17 @@ projects into the Worker: `team.members[]` with normalization as `_roomflow_room_meta`). Empty result → notification skipped with a `skipped` reason (standalone runs). +### Human resolution (@initiator) + +`_team_human_matrix_ids()` reads the same roster: every member whose +normalized role is neither leader nor worker, plus the `team.admin` +entry. These are the human users of the team (the task initiator +included) and are mentioned alongside the leader in completion and +attention events. Humans are not membership-checked (only the leader +is): a human who is not in the room simply cannot see the room event, +which is the correct Matrix semantics. Empty list → leader-only +mentions (standalone runs unchanged). + ### Membership guard `_validate_assignee_membership(room_id, leader)` is reused as-is: when @@ -78,31 +119,97 @@ the Matrix env is configured, the leader must be a joined member of the task room; otherwise the send is skipped (reason recorded) instead of producing an error event for a user who cannot receive it. -### Idempotency +### Idempotency (status-scoped, v2) + +- Stable txn `submit--`: Matrix de-duplicates a + redelivered identical PUT for the *same status*. +- `completionEventId` **and** `completionEventStatus` are persisted in + task state after the first success. A later resubmit with the **same** + status returns the recorded event with `reused: true` and performs no + HTTP call. A resubmit with a **changed** status invalidates the + recorded pair and sends a fresh event (different txn), so a worker + that first reports `BLOCKED` and later `SUCCESS` wakes the leader + again instead of being silently absorbed by the reuse branch. +- Tasks recorded before the upgrade (no `completionEventStatus`) keep + the old behavior: any resubmit reuses the recorded event. + +### P0 ordering: sync first, then notify (v2) + +The completion event is an *attention signal*, not a receipt. The +submit sequence is: local state → publish artifacts → **sync shared +storage → only then notify**. + +- **Sync failure** → `ok: false`, `retryable: true`, **no notification + at all** (the field is withheld from the response). The local task + state is already `submitted`, so the retry is idempotent: the event + is sent exactly once, on the first sync that succeeds. This closes + the "leader told done, artifacts unreachable" window — the leader + cannot be woken by an event whose artifacts it cannot read. +- **Notification-level failure** (no room, no leader, no Matrix env, + membership missing, HTTP error) stays best-effort: it returns + `{"sent": false, "skipped"?: true, "error": "..."}` and the submit + still completes: `ok: true`, `status: "submitted"`, artifacts + published, state synced. +- The existing `notificationNeeded` hint is **kept** — it also drives + the requester reply-route report, which the code-level line + intentionally does not cover (different room/audience). + +## Lifecycle attention events (v2, issue #1229) -- Stable txn `submit-`: Matrix de-duplicates a redelivered - identical PUT. -- `completionEventId` persisted in task state after the first success: - a later resubmit returns the recorded event with `reused: true` and - performs no HTTP call. +### `request_attention` (new taskflow action) -### Best-effort by contract +In-flight human decisions are currently "ambient room chat" — a worker +that needs approval/decision/escalation pokes the group and hopes a +human notices. The new action makes that a first-class, idempotent, +auditable event: -The completion message is a *notification*, not part of the state -mutation. Any problem (no room, no leader, no Matrix env, membership -missing, HTTP error) returns -`{"sent": false, "skipped"?: true, "error": "..."}` and the submit -still completes: `ok: true`, `status: "submitted"`, artifacts -published, state synced. The existing `notificationNeeded` hint is -**kept** — it also drives the requester reply-route report, which the -code-level line intentionally does not cover (different room/audience). +- **Roles**: worker / leader / remote-member. Terminal tasks are + rejected (`_require_task_mutable`). +- **Payload**: `kind` ∈ `approval | decision | escalation | other`, + `question` (required, ≤500 chars), optional `resolved: true` to + close it without a result. +- **Contract line**: `@leader ATTENTION_: - ` + + `- Worker:` line; mentions leader + human members. +- **State**: appends an `attention` record to task meta + (`kind / question / attempt / requestedAt / resolved / eventId?`). + Re-requesting the same `kind` while an unresolved record exists + reuses the recorded event (no duplicate ping); a new kind or a new + attempt number gets a fresh event (txn + `attention---`). +- **Sync-first** like submit: a failed sync withholds the notification + and returns a retryable failure. +- **Resolution**: `accept_task_result` marks **all** unresolved + attention records on the task `resolved: true` (the leader's decision + closed the loop), or an explicit `resolved: true` call closes a + record early. + +Routing-salience note: v1 delivers all attention in the task room +(room @mentions). A dedicated DM step for humans (higher salience) is +recorded as a follow-up in the PR, not in this change. + +### `PROJECT_COMPLETED` on `complete_project` (v2) + +`complete_project` previously only wrote state; a finished project +waited for the next incident to surface. It now sends a best-effort +room event before the state write: + +- **Contract line**: `@leader PROJECT_COMPLETED: - + Project completed: `; mentions leader + human members. +- Room resolution: first task `room_id` in the plan, falling back to + the project `source_room_id` when it is a Matrix room. +- **Idempotent**: `projectCompletionEventId` is persisted on the + project state (txn `project-<project-id>-success`); a retried + `complete_project` reuses the recorded event. +- Never blocks the terminal project write (same best-effort guards as + completion events). ## Changes | File | Change | |:--|:--| -| `plugins/teamharness/mcp/server.py` | `_team_leader_matrix_id()`, `_send_task_completion_notification()`, `_task_completion_notification()`; `submit_task` branch calls the orchestrator after `_publish_task_artifacts` and adds `notification` to the response | -| `plugins/tests/teamharness/mcp/tools/test-taskflow.rb` | runtime config gains the team roster; fake Matrix server gains a `submit-` fault-injection branch; new assertions (see below); context file-event selection made mxcUri-based instead of positional (the last event is no longer guaranteed to be a file event) | +| `plugins/teamharness/mcp/server.py` | v1: `_team_leader_matrix_id()`, `_send_task_completion_notification()`, `_task_completion_notification()`; `submit_task` branch adds `notification` to the response. v2: `_TASK_COMPLETION_EVENT_TOKENS` + per-status first-line rendering + status-scoped txn + `completionEventStatus` (status-scoped reuse); `_team_human_matrix_ids()` @initiator mentions; `submit_task` status validation + sync-before-notify ordering (retryable failure withholds the notification); new `request_attention` action + `_send_attention_notification()` (idempotent per kind, terminal guard, sync-first); `accept_task_result` auto-resolves outstanding attention; `_send_project_completion_notification()` + idempotent `projectCompletionEventId` on `complete_project` | +| `plugins/teamharness/skills/team/task-execution/SKILL.md` | contract section rewritten: code-generated per-status event lines (worker no longer hand-sends the completion line), status list extended to the full accepted set, `request_attention` documented as the in-flight decision path | +| `plugins/tests/teamharness/mcp/tools/test-taskflow.rb` | runtime config gains the team roster; fake Matrix server gains a `submit-` fault-injection branch; `mc` shim gains a `TEAMHARNESS_TEST_FAIL_SYNC_TASK` hook; new assertions (see below); context file-event selection made mxcUri-based instead of positional (the last event is no longer guaranteed to be a file event) | ## Tests (contract, `test-taskflow.rb`) @@ -120,6 +227,31 @@ code-level line intentionally does not cover (different room/audience). `ok: true` / `submitted`, `notification.sent: false` with the HTTP error, no `completionEventId` persisted. +v2 additions (issue #1229): + +6. **P0 ordering**: `mc` shim forced to fail for one task → submit + returns `ok: false` / `retryable: true` with **no `notification` + field**, local state still `submitted`; the idempotent retry after + storage recovery sends the event exactly once. +7. **Per-status token + @initiator**: `PARTIAL` / `FAILED` / + `REVISION_NEEDED` each render their own first-line token + + `- Status:` line, and the event mentions both the leader and the + human roster member; `SUCCESS` keeps the `Result:` line and carries + no `- Status:` line. +8. **Status validation**: submit with an unknown status is rejected + (`invalid status`) and the bad value is not persisted. +9. **Changed-status resubmit**: `FAILED` → `SUCCESS` resubmit sends a + second, distinct event (no silent reuse); a same-status resubmit + reuses the recorded event. +10. **request_attention**: in-flight `approval` ping sends the + `ATTENTION_APPROVAL` line with leader + human mentions; an + unresolved same-kind repeat is idempotent (no second event); a + different kind is not reused; a terminal (cancelled) task is + rejected; `accept_task_result` resolves the outstanding records. +11. **PROJECT_COMPLETED**: `complete_project` sends the + `PROJECT_COMPLETED` line with leader + human mentions; a retried + `complete_project` reuses the recorded event. + ## Open questions 1. **Should `check_task`'s polling be removed from leader prompts?** The diff --git a/plugins/teamharness/mcp/server.py b/plugins/teamharness/mcp/server.py index a2a03eda7..686b2273b 100644 --- a/plugins/teamharness/mcp/server.py +++ b/plugins/teamharness/mcp/server.py @@ -462,8 +462,10 @@ "description": ( "Coordinate bounded TeamHarness tasks after a project node is ready: " "leader delegates and checks tasks; worker or remote-member " - "acknowledges and submits results. Do not use for direct questions, " - "readiness checks, or ordinary conversation." + "acknowledges and submits results; worker or leader raises in-flight " + "attention (approval / decision / escalation) with request_attention. " + "Do not use for direct questions, readiness checks, or ordinary " + "conversation." ), "inputSchema": { "type": "object", @@ -475,8 +477,19 @@ }, "action": { "type": "string", - "enum": ["delegate_task", "ack_task", "submit_task", "check_task", "cancel_task"], - "description": "Task lifecycle operation.", + "enum": [ + "delegate_task", + "ack_task", + "submit_task", + "check_task", + "cancel_task", + "request_attention", + ], + "description": ( + "Task lifecycle operation. request_attention pulls a human " + "decision (kind: approval / decision / escalation / other) " + "out of the group chat while the task is still in flight." + ), }, "projectId": { "type": "string", @@ -3098,6 +3111,20 @@ def _accept_task_result(arguments: dict[str, Any], payload: dict[str, Any]) -> d break if not changed: raise ValueError("task not found in project plan") + # Accepting the result resolves ALL outstanding attention requests + # on the task: the leader's decision closed the loop, so no room + # ping on this task counts as an open loop anymore. + task_state = _read_json(_task_state_path(arguments, task_id), {}) + if isinstance(task_state, dict) and task_state: + attention = task_state.get("attention") + if isinstance(attention, list): + attention_changed = False + for item in attention: + if isinstance(item, dict) and not item.get("resolved"): + item["resolved"] = True + attention_changed = True + if attention_changed: + _write_task(arguments, task_state) result_status = str(result_status_value or "SUCCESS") if node_status == "completed": project["requester_report"] = { @@ -3249,6 +3276,7 @@ def _project_id_for_pull(arguments: dict[str, Any], payload: dict[str, Any]) -> "ack_task", "submit_task", "cancel_task", + "request_attention", }) @@ -3571,6 +3599,17 @@ def _projectflow(arguments: dict[str, Any]) -> dict[str, Any]: loop["status"] = "completed" project["loop"] = loop project_dir = _project_dir(arguments, project_id) + if action == "complete_project": + # Code-level PROJECT_COMPLETED attention event (PR review + # 2026-09-05): the leader and the human members see a + # finished project in the room instead of waiting for the + # next incident. Best-effort — it never blocks the + # terminal project write. Runs before the state write so + # a recorded projectCompletionEventId makes a retried + # complete_project idempotent. + project["projectNotification"] = _send_project_completion_notification( + arguments, project, project_id + ) _write_json(state_path, project) _write_project_plan(project_dir, project) _sync_project(arguments, project_id) @@ -4101,8 +4140,122 @@ def _team_leader_matrix_id() -> str: return "" -def _send_task_completion_notification( +_TASK_COMPLETION_EVENT_TOKENS = { + "SUCCESS": "TASK_COMPLETED", + "SUCCESS_WITH_NOTES": "TASK_COMPLETED", + "PARTIAL": "TASK_PARTIAL", + "REVISION_NEEDED": "TASK_REVISION_NEEDED", + "BLOCKED": "TASK_BLOCKED", + "FAILED": "TASK_FAILED", + # Defensive only — not part of the accepted result-status contract. + "INTERRUPTED": "TASK_INTERRUPTED", +} + + +def _team_human_matrix_ids() -> list[str]: + """Resolve Matrix IDs of human (non-leader, non-worker) team members. + + Same runtime-config source as ``_team_leader_matrix_id``: the + controller projects the full roster (including humans and the team + admin) into ``team.members`` with their roles. Humans are every + member whose role is neither leader nor worker, plus the + ``team.admin`` entry. Returns an empty list on standalone runs so + callers skip the extra mentions instead of failing. + """ + config = _load_runtime_config() + team = _section(config, "team") + ids: list[str] = [] + + def _add(user_id: Any) -> None: + user_id = str(user_id or "").strip() + if user_id.startswith("@") and user_id not in ids: + ids.append(user_id) + + admin = _section(team, "admin") + _add(admin.get("matrixUserId") or admin.get("matrix_user_id")) + members = team.get("members") + if not isinstance(members, list): + return ids + for member in members: + if not isinstance(member, dict): + continue + role = str(member.get("role") or "").strip().lower().replace("_", "-") + if role in {"team-leader", "teamleader", "leader", "worker"}: + continue + _add(member.get("matrixUserId") or member.get("matrix_user_id")) + return ids + + +def _send_project_completion_notification( arguments: dict[str, Any], + project: dict[str, Any], + project_id: str, +) -> dict[str, Any]: + """Send the automatic PROJECT_COMPLETED event for complete_project. + + Best-effort: every guard failure returns a ``skipped`` result and + never blocks the terminal project write. The event mentions the team + leader and the human members in the task room so the requester sees + project completion with the same salience as a task completion. The + recorded event id makes a retried complete_project idempotent. + """ + leader = _team_leader_matrix_id() + if not leader: + return { + "sent": False, + "skipped": True, + "error": "team leader Matrix ID not found in runtime config", + } + if project.get("projectCompletionEventId"): + return { + "sent": True, + "eventId": str(project["projectCompletionEventId"]), + "leader": leader, + "reused": True, + } + room_id = "" + tasks = project.get("tasks") + if isinstance(tasks, list): + for task in tasks: + if not isinstance(task, dict): + continue + room_id = str(task.get("room_id") or "").strip() + if not room_id: + task_id = str(task.get("task_id") or task.get("taskId") or "").strip() + if task_id: + task_state = _read_json(_task_state_path(arguments, task_id), {}) + room_id = str(task_state.get("room_id") or "").strip() + if room_id: + break + if not room_id: + source_room_id = str(project.get("source_room_id") or "").strip() + if MATRIX_ROOM_RE.fullmatch(_canonical_room_id(source_room_id)): + room_id = source_room_id + if not room_id: + return {"sent": False, "skipped": True, "error": "project has no room_id"} + membership = _validate_assignee_membership(room_id, leader) + if not membership.get("ok"): + return { + "sent": False, + "skipped": True, + "error": str(membership.get("error") or "room membership check failed"), + } + notification = _send_task_completion_notification( + room_id=room_id, + task_id=project_id, + status="SUCCESS", + summary=f"Project completed: {str(project.get('title') or project_id)[:200]}", + leader=leader, + humans=_team_human_matrix_ids(), + event_token="PROJECT_COMPLETED", + txn_prefix="project", + ) + if notification.get("sent"): + project["projectCompletionEventId"] = notification.get("eventId") + return notification + + +def _send_task_completion_notification( *, room_id: str, task_id: str, @@ -4111,17 +4264,27 @@ def _send_task_completion_notification( leader: str, worker: str = "", result_path: str = "", + humans: list[str] | None = None, + event_token: str = "", + txn_prefix: str = "submit", ) -> dict[str, Any]: """Send the automatic Worker completion notification for submit_task. - Publishes the completion line to the Task room with ``m.mentions`` - using the same Matrix HTTP send path as the message tool. The first - line follows the task-execution skill contract so leader-side prompts - that parse completion lines keep working: + Publishes the completion event to the Task room with ``m.mentions`` + using the same Matrix HTTP send path as the message tool. Every + result status gets its own first-line contract token so leader-side + prompts can branch on the line itself (task-execution skill + contract): @leader TASK_COMPLETED: <task-id> - Result: shared/tasks/<task-id>/result.md - @leader BLOCKED: <task-id> - <short blocker summary> - The transaction ID is stable per task so a retry cannot produce a - duplicate completion. + @leader TASK_PARTIAL: <task-id> - <summary> + @leader TASK_REVISION_NEEDED: <task-id> - <summary> + @leader TASK_BLOCKED: <task-id> - <short blocker summary> + @leader TASK_FAILED: <task-id> - <summary> + Humans (task initiator and other non-agent members) are mentioned + alongside the leader so the requester is routed with the same + salience the leader is. The transaction ID is stable per task and + status so a retry cannot produce a duplicate event, while a + re-submission with a changed status produces a new one. """ homeserver = os.getenv("AGENTTEAMS_MATRIX_URL", "").rstrip("/") token = os.getenv("AGENTTEAMS_WORKER_MATRIX_TOKEN", "") @@ -4140,25 +4303,121 @@ def _send_task_completion_notification( summary_preview = (summary or "")[:500] if len(summary or "") > 500: summary_preview += "..." - if status == "BLOCKED": - notification_text = f"{leader} BLOCKED: {task_id} - {summary_preview}" - detail = "" - else: + line_token = event_token or _TASK_COMPLETION_EVENT_TOKENS.get(status, f"TASK_{status}") + if event_token: + notification_text = f"{leader} {event_token}: {task_id} - {summary_preview}".rstrip() + elif line_token == "TASK_COMPLETED": if result_path: notification_text = f"{leader} TASK_COMPLETED: {task_id} - Result: {result_path}" - else: + elif summary_preview: notification_text = f"{leader} TASK_COMPLETED: {task_id} - {summary_preview}" - detail = f"\n{summary_preview}" if summary_preview else "" + else: + notification_text = f"{leader} TASK_COMPLETED: {task_id}" + else: + notification_text = f"{leader} {line_token}: {task_id} - {summary_preview}".rstrip() if worker: notification_text += f"\n- Worker: {worker}" - if status in {"REVISION_NEEDED", "INTERRUPTED"}: + if not event_token and line_token != "TASK_COMPLETED": notification_text += f"\n- Status: {status}" - notification_text += detail + elif ( + line_token == "TASK_COMPLETED" + and not event_token + and result_path + and summary_preview + ): + # The Result line names the file; keep the short summary preview + # as the detail line (legacy contract, leader-side prompts read it). + notification_text += f"\n{summary_preview}" + mentions = [leader] + for human in humans or []: + if human and human != leader and human not in mentions: + mentions.append(human) + content = _matrix_content(notification_text, mentions) + + room_enc = urllib.parse.quote(matrix_room_id, safe="") + txn = urllib.parse.quote(f"{txn_prefix}-{task_id}-{status.lower()}", safe="") + url = f"{homeserver}/_matrix/client/v3/rooms/{room_enc}/send/m.room.message/{txn}" + request = urllib.request.Request( + url, + data=json.dumps(content).encode("utf-8"), + headers={ + "Authorization": f"Bearer {token}", + "Content-Type": "application/json", + }, + method="PUT", + ) + try: + with urllib.request.urlopen(request, timeout=10) as response: + data = json.loads(response.read().decode("utf-8") or "{}") + event_id = str(data.get("event_id") or "").strip() + if not event_id: + return {"sent": False, "error": "Matrix send returned no event_id"} + return { + "sent": True, + "eventId": event_id, + "roomId": matrix_room_id, + "leader": leader, + "humans": mentions[1:], + } + except urllib.error.HTTPError as exc: + body = exc.read().decode("utf-8", errors="replace")[:200] + return { + "sent": False, + "error": f"Matrix API error: HTTP {exc.code}: {body}", + } + except (urllib.error.URLError, TimeoutError, OSError) as exc: + return {"sent": False, "error": f"Matrix API error: {exc}"} + + +def _send_attention_notification( + arguments: dict[str, Any], + *, + room_id: str, + task_id: str, + kind: str, + question: str, + leader: str, + worker: str = "", + humans: list[str] | None = None, + attempt: int = 1, +) -> dict[str, Any]: + """Send an in-flight attention request to the Task room. + + Workers/leaders use this to pull a human decision (approval, + decision, escalation) out of the ambient group chat: the event is a + status-scoped first-line token (``ATTENTION_APPROVAL`` etc.) that + mentions the leader and the human members of the team. The + transaction ID is stable per task, kind and attempt so a retry + cannot duplicate the event while a new attempt produces a new one. + """ + homeserver = os.getenv("AGENTTEAMS_MATRIX_URL", "").rstrip("/") + token = os.getenv("AGENTTEAMS_WORKER_MATRIX_TOKEN", "") + if not homeserver or not token: + return { + "sent": False, + "error": "AGENTTEAMS_MATRIX_URL and AGENTTEAMS_WORKER_MATRIX_TOKEN are required", + } + + matrix_room_id = str(room_id or "").strip() + if matrix_room_id.startswith("room:"): + matrix_room_id = matrix_room_id[len("room:") :].strip() + if not matrix_room_id.startswith("!"): + return {"sent": False, "error": f"invalid Matrix room target: {room_id}"} + + question_preview = (question or "")[:500] + if len(question or "") > 500: + question_preview += "..." + notification_text = f"{leader} ATTENTION_{kind.upper()}: {task_id} - {question_preview}".rstrip() + if worker: + notification_text += f"\n- Worker: {worker}" mentions = [leader] + for human in humans or []: + if human and human != leader and human not in mentions: + mentions.append(human) content = _matrix_content(notification_text, mentions) room_enc = urllib.parse.quote(matrix_room_id, safe="") - txn = urllib.parse.quote(f"submit-{task_id}", safe="") + txn = urllib.parse.quote(f"attention-{task_id}-{kind}-{attempt}", safe="") url = f"{homeserver}/_matrix/client/v3/rooms/{room_enc}/send/m.room.message/{txn}" request = urllib.request.Request( url, @@ -4180,6 +4439,7 @@ def _send_task_completion_notification( "eventId": event_id, "roomId": matrix_room_id, "leader": leader, + "humans": mentions[1:], } except urllib.error.HTTPError as exc: body = exc.read().decode("utf-8", errors="replace")[:200] @@ -4216,14 +4476,24 @@ def _task_completion_notification( "skipped": True, "error": "team leader Matrix ID not found in runtime config", } + recorded_status = str(task.get("completionEventStatus") or "") if task.get("completionEventId"): - return { - "sent": True, - "eventId": str(task["completionEventId"]), - "roomId": _canonical_room_id(room_id), - "leader": leader, - "reused": True, - } + if not recorded_status or recorded_status == status: + return { + "sent": True, + "eventId": str(task["completionEventId"]), + "roomId": _canonical_room_id(room_id), + "leader": leader, + "reused": True, + "completionStatus": recorded_status or status, + } + # A recorded completion exists but the task is being re-submitted + # with a changed result status: the recorded event is stale. + # Invalidate it so the new status-scoped transaction can send a + # fresh event instead of silently reusing the old one. + task.pop("completionEventId", None) + task.pop("completionEventStatus", None) + _write_task(arguments, task) membership = _validate_assignee_membership(room_id, leader) if not membership.get("ok"): return { @@ -4232,7 +4502,6 @@ def _task_completion_notification( "error": str(membership.get("error") or "room membership check failed"), } notification = _send_task_completion_notification( - arguments, room_id=room_id, task_id=task_id, status=status, @@ -4240,9 +4509,15 @@ def _task_completion_notification( leader=leader, worker=str(task.get("assigned_to") or ""), result_path=str(task.get("result_path") or ""), + humans=_team_human_matrix_ids(), ) if notification.get("sent"): + # If the persist below fails, the event id is lost but the + # transaction id is stable per task + status, so a retried + # submission re-puts the same Matrix txn and the homeserver + # deduplicates it — no duplicate event, the id is re-recorded. task["completionEventId"] = notification.get("eventId") + task["completionEventStatus"] = status _write_task(arguments, task) return notification @@ -4536,6 +4811,11 @@ def _taskflow(arguments: dict[str, Any]) -> dict[str, Any]: _require_task_mutable(arguments, task, task_id, action) summary = str(payload.get("summary") or "") status = str(payload.get("status") or "SUCCESS") + if status not in ALLOWED_TASK_RESULT_STATUSES: + raise ValueError( + f"invalid status {status!r}; expected one of " + f"{', '.join(sorted(ALLOWED_TASK_RESULT_STATUSES))}" + ) deliverables = payload.get("deliverables") or [] if not isinstance(deliverables, list): raise ValueError("deliverables must be a list") @@ -4562,6 +4842,29 @@ def _taskflow(arguments: dict[str, Any]) -> dict[str, Any]: deliverables, _attachment_parent_event_id(payload, arguments), ) + # P0 ordering (PR review 2026-09-05): the completion event is + # an attention signal, not a receipt. Sync shared storage + # first so the leader can immediately read result.md / + # deliverables, and only then notify. A failed sync withholds + # the notification and returns a retryable failure instead of + # telling the leader "done" while the artifacts are + # unreachable. The retry is idempotent (event reuse is keyed + # by task + status). + synced = _sync_task(arguments, task_id, exclude=["spec.md", "base/"]) + if not synced: + return { + "ok": False, + "retryable": True, + "tool": "taskflow", + "action": action, + "task": task, + "error": ( + "shared storage sync failed after submit; the completion " + "notification was withheld. Local task state is already " + "submitted — retry submit_task (idempotent) once storage " + "recovers." + ), + } notification = _task_completion_notification( arguments, task, @@ -4575,7 +4878,7 @@ def _taskflow(arguments: dict[str, Any]) -> dict[str, Any]: "action": action, "task": task, "publishedArtifacts": published_artifacts, - "synced": _sync_task(arguments, task_id, exclude=["spec.md", "base/"]), + "synced": True, "notification": notification, "notificationNeeded": _notification_needed( "submit_task", @@ -4585,6 +4888,150 @@ def _taskflow(arguments: dict[str, Any]) -> dict[str, Any]: ), } + if action == "request_attention": + if role not in {"worker", "leader", "remote-member"}: + raise ValueError( + "request_attention requires worker, leader, or remote-member role" + ) + task_id = _safe_id(payload.get("taskId") or payload.get("task_id"), "taskId") + task = _load_task(arguments, task_id) + _require_task_mutable(arguments, task, task_id, action) + kind = str(payload.get("kind") or "other").strip().lower() + if kind not in {"approval", "decision", "escalation", "other"}: + raise ValueError("kind must be one of approval, decision, escalation, other") + question = str(payload.get("question") or payload.get("reason") or "").strip() + if not question: + raise ValueError("question is required") + attention = task.get("attention") + if not isinstance(attention, list): + attention = [] + # Idempotent per kind while unresolved: re-requesting the same + # kind before it was resolved reuses the recorded event instead + # of pinging the room again. An explicit resolved=true closes + # the open loop early (no new ping). + explicit_resolve = bool(payload.get("resolved", False)) + for existing in reversed(attention): + if ( + isinstance(existing, dict) + and not existing.get("resolved") + and str(existing.get("kind") or "") == kind + and existing.get("eventId") + ): + if explicit_resolve: + existing["resolved"] = True + _write_task(arguments, task) + return { + "ok": True, + "tool": "taskflow", + "action": action, + "task": task, + "attention": { + "kind": kind, + "resolved": True, + "eventId": str(existing["eventId"]), + "notification": { + "sent": True, + "eventId": str(existing["eventId"]), + "reused": True, + "resolved": True, + }, + }, + "synced": True, + } + return { + "ok": True, + "tool": "taskflow", + "action": action, + "task": task, + "attention": { + "kind": kind, + "reused": True, + "notification": { + "sent": True, + "eventId": str(existing["eventId"]), + "reused": True, + }, + }, + "synced": True, + } + attempt = sum( + 1 for item in attention if isinstance(item, dict) and item.get("kind") == kind + ) + 1 + record: dict[str, Any] = { + "kind": kind, + "question": question[:500], + "attempt": attempt, + "requestedAt": datetime.datetime.now(datetime.timezone.utc).isoformat(), + "resolved": bool(payload.get("resolved", False)), + } + attention.append(record) + task["attention"] = attention + _write_task(arguments, task) + synced = _sync_task(arguments, task_id, exclude=["spec.md", "base/"]) + if not synced: + return { + "ok": False, + "retryable": True, + "tool": "taskflow", + "action": action, + "task": task, + "error": ( + "shared storage sync failed after request_attention; the " + "attention notification was withheld. Local attention state " + "is recorded — retry request_attention (idempotent)." + ), + } + room_id = str(task.get("room_id") or "").strip() + if not room_id: + notification = { + "sent": False, + "skipped": True, + "error": "task has no room_id", + } + else: + leader = _team_leader_matrix_id() + if not leader: + notification = { + "sent": False, + "skipped": True, + "error": "team leader Matrix ID not found in runtime config", + } + else: + membership = _validate_assignee_membership(room_id, leader) + if not membership.get("ok"): + notification = { + "sent": False, + "skipped": True, + "error": str(membership.get("error") or "room membership check failed"), + } + else: + notification = _send_attention_notification( + arguments, + room_id=room_id, + task_id=task_id, + kind=kind, + question=question, + leader=leader, + worker=str(task.get("assigned_to") or ""), + humans=_team_human_matrix_ids(), + attempt=attempt, + ) + if notification.get("sent"): + record["eventId"] = notification.get("eventId") + _write_task(arguments, task) + return { + "ok": True, + "tool": "taskflow", + "action": action, + "task": task, + "attention": { + "kind": kind, + "attempt": attempt, + "notification": notification, + }, + "synced": True, + } + if action == "cancel_task": if role != "leader": raise ValueError("cancel_task requires leader role") diff --git a/plugins/teamharness/skills/team/task-execution/SKILL.md b/plugins/teamharness/skills/team/task-execution/SKILL.md index 844c02ecd..05bf412d2 100644 --- a/plugins/teamharness/skills/team/task-execution/SKILL.md +++ b/plugins/teamharness/skills/team/task-execution/SKILL.md @@ -120,8 +120,10 @@ Use one of: - `SUCCESS` - `SUCCESS_WITH_NOTES` +- `PARTIAL` - `REVISION_NEEDED` - `BLOCKED` +- `FAILED` Submitting ends the task. Do not keep editing the old task after submission unless the Leader assigns a new task. @@ -139,21 +141,28 @@ file panel. ## Completion Message -After `submit_task` returns `ok: true`, send a normal text message in the -current Task room and mention the Leader with the exact Matrix user id or -resolvable mention from the task spec: +`submit_task` automatically publishes the completion event to the Task room: +the first line is the contract below, and the event @mentions the Leader +and the human members of the team (the task initiator), so the requester is +routed with the same salience the leader is. After `submit_task` returns +`ok: true`, do not send another completion line. + +The event first line carries one token per result status (code-generated): ```text @leader-user:matrix.local TASK_COMPLETED: demo-project-001-01 - Result: shared/tasks/demo-project-001-01/result.md +@leader-user:matrix.local TASK_PARTIAL: demo-project-001-01 - <summary> +@leader-user:matrix.local TASK_REVISION_NEEDED: demo-project-001-01 - <summary> +@leader-user:matrix.local TASK_BLOCKED: demo-project-001-01 - <short blocker summary> +@leader-user:matrix.local TASK_FAILED: demo-project-001-01 - <summary> ``` -If the task spec gives an exact completion line, preserve that line exactly and -include one short summary sentence. A tool call, tool-output thread, or -`result.md` file does not count as the completion message. Do not use -`NO_REPLY` after successful submission. - -For blockers: +If the task spec gives an exact completion line, the code event keeps that +line format; a short human-readable summary message may still follow in the +room but never replaces the event. Do not use `NO_REPLY` after successful +submission. -```text -@leader-user:matrix.local BLOCKED: demo-project-001-01 - <short blocker summary> -``` +While the task is still in flight and you need a human decision (approval / +decision / escalation) instead of guessing, call `taskflow` with +`action: request_attention` (payload: `kind`, `question`). Do not rely on +the human noticing an ambient room message. diff --git a/plugins/tests/teamharness/mcp/tools/test-taskflow.rb b/plugins/tests/teamharness/mcp/tools/test-taskflow.rb index 09d50f0b9..bed19a64c 100644 --- a/plugins/tests/teamharness/mcp/tools/test-taskflow.rb +++ b/plugins/tests/teamharness/mcp/tools/test-taskflow.rb @@ -36,6 +36,12 @@ def fail!(message) (bin_dir / "mc").write(<<~SH) #!/usr/bin/env bash printf '%s\\n' "$*" >> "#{log_path}" + # Test hook: fail the push (mirror <local> <remote>) for the named + # task only — pre-action pulls keep working (used to exercise the + # submit sync-failure withholding path). + if [ "$1" = "mirror" ] && [ -n "${TEAMHARNESS_TEST_FAIL_SYNC_TASK:-}" ] && [ "$3" = "mock/shared/tasks/${TEAMHARNESS_TEST_FAIL_SYNC_TASK}/" ]; then + exit 1 + fi if [ "$1" = "mirror" ] && [ "$2" = "mock/shared/tasks/remote-001/" ]; then mkdir -p "$3" cp -a "#{remote_task}/." "$3" @@ -131,6 +137,8 @@ def do_GET(self): {"state_key": "@worker-invited:example.test", "content": {"membership": "invite"}}, {"state_key": "@admin:example.test", "content": {"membership": "join"}}, ] + if os.environ.get("TEAMHARNESS_TEST_EXCLUDE_LEADER_FROM_ROOM") == "1": + members = [m for m in members if m["state_key"] != "@admin:example.test"] payload = {"chunk": members} self.send_response(200) self.send_header("Content-Type", "application/json") @@ -1337,6 +1345,310 @@ def completion_events(): if final_meta.get("assigned_at") != assigned_at: raise AssertionError(f"final task meta should preserve assigned_at: {final_meta!r}") + # ================================================================== + # PR review 2026-09-05 (issue #1229): lifecycle attention events + # ================================================================== + + # A human member (task initiator) joins the roster so @initiator + # routing can be asserted. + runtime_cfg = pathlib.Path("#{root}") / "runtime.yaml" + runtime_cfg.write_text( + runtime_cfg.read_text(encoding="utf-8").rstrip() + + "\\n - name: 'Luo'\\n" + " runtimeName: 'luo'\\n" + " role: 'human'\\n" + " matrixUserId: '@luo:example.test'\\n", + encoding="utf-8", + ) + + def _lifecycle_setup(tid): + pid = f"attn-{tid}" + payload("projectflow", { + "action": "create_project", + "payload": {"projectId": pid, "title": "Attention fixture"}, + }) + payload("projectflow", { + "action": "plan_dag", + "payload": {"projectId": pid, "tasks": [{ + "taskId": tid, + "title": "Attention task", + "assignedTo": "@worker-a:example.test", + "dependsOn": [], + }]}, + }) + delegated = payload("taskflow", { + "role": "leader", + "action": "delegate_task", + "payload": { + "projectId": pid, + "taskId": tid, + "roomId": "room:!team:example.test", + "spec": "Attention spec.", + }, + }) + if not delegated.get("ok"): + raise AssertionError(f"delegate_task failed for {tid}: {delegated!r}") + acked = payload("taskflow", { + "role": "worker", + "action": "ack_task", + "payload": {"taskId": tid}, + }) + if not acked.get("ok"): + raise AssertionError(f"ack_task failed for {tid}: {acked!r}") + tdir = pathlib.Path("#{workspace}") / f"shared/tasks/{tid}" + tdir.mkdir(parents=True, exist_ok=True) + (tdir / "result.md").write_text("Result body\\n", encoding="utf-8") + return pid + + def _lifecycle_submit(tid, status, summary="Done."): + return payload("taskflow", { + "role": "worker", + "action": "submit_task", + "payload": {"taskId": tid, "status": status, "summary": summary}, + }) + + # --- P0 ordering: a failed shared-storage sync withholds the + # completion notification and returns a retryable failure; the + # idempotent retry delivers it once storage recovers. --- + _lifecycle_setup("order-task") + os.environ["TEAMHARNESS_TEST_FAIL_SYNC_TASK"] = "order-task" + try: + order_result = _lifecycle_submit("order-task", "SUCCESS", "Result ready but storage is down.") + finally: + os.environ.pop("TEAMHARNESS_TEST_FAIL_SYNC_TASK", None) + if order_result.get("ok") is not False: + raise AssertionError(f"failed sync must not report ok: {order_result!r}") + if order_result.get("retryable") is not True: + raise AssertionError(f"failed sync must be retryable: {order_result!r}") + if "notification" in order_result: + raise AssertionError(f"failed sync must withhold the completion notification: {order_result!r}") + if not order_result.get("task") or order_result["task"]["status"] != "submitted": + raise AssertionError(f"local task state must still be submitted: {order_result!r}") + order_retry = _lifecycle_submit("order-task", "SUCCESS", "Result ready but storage is down.") + if not order_retry.get("ok") or order_retry.get("synced") is not True: + raise AssertionError(f"retry after storage recovery must succeed: {order_retry!r}") + if (order_retry.get("notification") or {}).get("sent") is not True: + raise AssertionError(f"retry after storage recovery must send the notification: {order_retry!r}") + + # --- Skip branches: the completion notification is best-effort — + # missing Matrix env, a missing leader, or a leader that is not a + # joined room member each skip the notification without blocking + # the submission itself. --- + _lifecycle_setup("skip-env-task") + saved_url = os.environ.pop("AGENTTEAMS_MATRIX_URL", None) + saved_token = os.environ.pop("AGENTTEAMS_WORKER_MATRIX_TOKEN", None) + try: + env_res = _lifecycle_submit("skip-env-task", "SUCCESS", "Matrix env unavailable.") + if not env_res.get("ok"): + raise AssertionError(f"missing matrix env must not block submit_task: {env_res!r}") + env_notif = env_res.get("notification") or {} + if env_notif.get("sent") is not False or "AGENTTEAMS_MATRIX_URL" not in str(env_notif.get("error", "")): + raise AssertionError(f"missing matrix env must skip with a clear error: {env_notif!r}") + finally: + if saved_url is not None: + os.environ["AGENTTEAMS_MATRIX_URL"] = saved_url + if saved_token is not None: + os.environ["AGENTTEAMS_WORKER_MATRIX_TOKEN"] = saved_token + + _lifecycle_setup("skip-leader-task") + rc_path = pathlib.Path(os.environ["TEAMHARNESS_RUNTIME_CONFIG"]) + saved_rc = rc_path.read_text(encoding="utf-8") + try: + rc_path.write_text(saved_rc.replace("matrixUserId: '@admin:example.test'", "matrixUserId: ''"), encoding="utf-8") + leader_res = _lifecycle_submit("skip-leader-task", "SUCCESS", "No leader configured.") + if not leader_res.get("ok"): + raise AssertionError(f"missing leader must not block submit_task: {leader_res!r}") + leader_notif = leader_res.get("notification") or {} + if leader_notif.get("sent") is not False or "leader" not in str(leader_notif.get("error", "")).lower(): + raise AssertionError(f"missing leader must skip with a clear error: {leader_notif!r}") + finally: + rc_path.write_text(saved_rc, encoding="utf-8") + + _lifecycle_setup("skip-membership-task") + os.environ["TEAMHARNESS_TEST_EXCLUDE_LEADER_FROM_ROOM"] = "1" + try: + member_res = _lifecycle_submit("skip-membership-task", "SUCCESS", "Leader outside the room.") + if not member_res.get("ok"): + raise AssertionError(f"leader not in room must not block submit_task: {member_res!r}") + member_notif = member_res.get("notification") or {} + if member_notif.get("sent") is not False or "not a joined member" not in str(member_notif.get("error", "")): + raise AssertionError(f"leader outside the room must skip on membership: {member_notif!r}") + finally: + os.environ.pop("TEAMHARNESS_TEST_EXCLUDE_LEADER_FROM_ROOM", None) + + # --- Per-status first-line token + @initiator human mention. --- + for status, token in ( + ("PARTIAL", "TASK_PARTIAL"), + ("FAILED", "TASK_FAILED"), + ("REVISION_NEEDED", "TASK_REVISION_NEEDED"), + ): + tid = f"tok-{status.lower()}" + _lifecycle_setup(tid) + res = _lifecycle_submit(tid, status, f"Status {status} case.") + if not res.get("ok"): + raise AssertionError(f"submit {status} failed: {res!r}") + evs = [ev for ev in matrix["events"] if f"submit-{tid}-" in ev["path"]] + if len(evs) != 1: + raise AssertionError(f"expected exactly one completion event for {tid}: {evs!r}") + body = evs[0]["content"].get("body", "") + if f"{token}: {tid} -" not in body: + raise AssertionError(f"{status} event must carry the {token} first line: {body!r}") + if f"- Status: {status}" not in body: + raise AssertionError(f"{status} event must carry the Status line: {body!r}") + mentions = (evs[0]["content"].get("m.mentions") or {}).get("user_ids", []) + if "@admin:example.test" not in mentions or "@luo:example.test" not in mentions: + raise AssertionError(f"{status} event must mention leader and human initiator: {mentions!r}") + ok_tid = "tok-success" + _lifecycle_setup(ok_tid) + ok_res = _lifecycle_submit(ok_tid, "SUCCESS", "All good.") + ok_ev = [ev for ev in matrix["events"] if f"submit-{ok_tid}-" in ev["path"]] + ok_body = ok_ev[0]["content"].get("body", "") if ok_ev else "" + if f"TASK_COMPLETED: {ok_tid} - Result: shared/tasks/{ok_tid}/result.md" not in ok_body: + raise AssertionError(f"SUCCESS event must keep the Result line: {ok_body!r}") + if "- Status:" in ok_body: + raise AssertionError(f"SUCCESS event must not carry a Status line: {ok_body!r}") + + # --- Invalid status is rejected at submit. --- + bad_tid = "tok-invalid" + _lifecycle_setup(bad_tid) + bad = payload("taskflow", { + "role": "worker", + "action": "submit_task", + "payload": {"taskId": bad_tid, "status": "MAYBE", "summary": "Not a real status."}, + }) + if bad.get("ok") or "invalid status" not in str(bad.get("error", "")): + raise AssertionError(f"submit_task must reject unknown statuses: {bad!r}") + bad_meta = json.loads( + (pathlib.Path("#{workspace}") / f"shared/tasks/{bad_tid}/meta.json").read_text(encoding="utf-8") + ) + if bad_meta.get("result_status") == "MAYBE": + raise AssertionError(f"rejected status must not be persisted: {bad_meta!r}") + + # --- Re-submission with a changed status sends a new event. --- + ch_tid = "tok-resubmit" + _lifecycle_setup(ch_tid) + first = _lifecycle_submit(ch_tid, "FAILED", "First pass failed.") + n1 = (first.get("notification") or {}).get("eventId") + if not first.get("ok") or (first.get("notification") or {}).get("sent") is not True or not n1: + raise AssertionError(f"first FAILED submit must notify: {first!r}") + second = _lifecycle_submit(ch_tid, "SUCCESS", "Fixed on resubmit.") + n2 = (second.get("notification") or {}).get("eventId") + if not second.get("ok") or (second.get("notification") or {}).get("sent") is not True or not n2 or n2 == n1: + raise AssertionError(f"changed-status resubmit must send a new event: {second!r} (first={n1}, second={n2})") + if len([ev for ev in matrix["events"] if f"submit-{ch_tid}-" in ev["path"]]) != 2: + raise AssertionError("changed-status resubmit must not reuse the old event") + same = _lifecycle_submit(ch_tid, "SUCCESS", "Idempotent same-status resubmit.") + if (same.get("notification") or {}).get("reused") is not True: + raise AssertionError(f"same-status resubmit must reuse the event: {same!r}") + if (same.get("notification") or {}).get("eventId") != n2: + raise AssertionError(f"same-status resubmit must reuse the same event id: {same!r}") + + # --- request_attention: in-flight ping, idempotent, terminal guard, + # resolved by accept_task_result. --- + att_tid = "att-task" + att_pid = _lifecycle_setup(att_tid) + att1 = payload("taskflow", { + "role": "worker", + "action": "request_attention", + "payload": {"taskId": att_tid, "kind": "approval", "question": "Ship to production?"}, + }) + if not att1.get("ok") or (att1.get("attention") or {}).get("notification", {}).get("sent") is not True: + raise AssertionError(f"request_attention must notify the room: {att1!r}") + att_ev = [ev for ev in matrix["events"] if f"attention-{att_tid}-approval-" in ev["path"]] + if len(att_ev) != 1: + raise AssertionError(f"expected one attention event: {att_ev!r}") + att_body = att_ev[0]["content"].get("body", "") + if f"ATTENTION_APPROVAL: {att_tid} - Ship to production?" not in att_body: + raise AssertionError(f"attention event must carry the contract line: {att_body!r}") + mentions = (att_ev[0]["content"].get("m.mentions") or {}).get("user_ids", []) + if "@admin:example.test" not in mentions or "@luo:example.test" not in mentions: + raise AssertionError(f"attention event must mention leader and human: {mentions!r}") + att2 = payload("taskflow", { + "role": "worker", + "action": "request_attention", + "payload": {"taskId": att_tid, "kind": "approval", "question": "Ship to production?"}, + }) + if not att2.get("ok") or (att2.get("attention") or {}).get("reused") is not True: + raise AssertionError(f"unresolved same-kind attention must be idempotent: {att2!r}") + if len([ev for ev in matrix["events"] if f"attention-{att_tid}-approval-" in ev["path"]]) != 1: + raise AssertionError("idempotent attention must not send a second event") + att3 = payload("taskflow", { + "role": "worker", + "action": "request_attention", + "payload": {"taskId": att_tid, "kind": "escalation", "question": "Escalating: storage at capacity."}, + }) + if not att3.get("ok") or (att3.get("attention") or {}).get("reused") is True: + raise AssertionError(f"different kind must not reuse the pending event: {att3!r}") + att_close = payload("taskflow", { + "role": "worker", + "action": "request_attention", + "payload": {"taskId": att_tid, "kind": "escalation", "question": "Closing early.", "resolved": True}, + }) + if not att_close.get("ok") or (att_close.get("attention") or {}).get("resolved") is not True: + raise AssertionError(f"explicit resolved=true must close the open loop: {att_close!r}") + _lifecycle_submit(att_tid, "BLOCKED", "Blocked on storage.") + accepted = payload("projectflow", { + "role": "leader", + "action": "accept_task_result", + "payload": {"projectId": att_pid, "taskId": att_tid, "resultStatus": "BLOCKED"}, + }) + if not accepted.get("ok"): + raise AssertionError(f"accept_task_result failed: {accepted!r}") + att_meta = json.loads( + (pathlib.Path("#{workspace}") / f"shared/tasks/{att_tid}/meta.json").read_text(encoding="utf-8") + ) + unresolved = [item for item in (att_meta.get("attention") or []) if not item.get("resolved")] + if unresolved: + raise AssertionError(f"accept_task_result must resolve outstanding attention: {att_meta.get('attention')!r}") + can_tid = "att-cancel" + can_pid = _lifecycle_setup(can_tid) + cancelled = payload("taskflow", { + "role": "leader", + "action": "cancel_task", + "payload": {"projectId": can_pid, "taskId": can_tid, "reason": "Cancelled for attention guard test."}, + }) + if not cancelled.get("ok"): + raise AssertionError(f"cancel_task failed: {cancelled!r}") + att5 = payload("taskflow", { + "role": "worker", + "action": "request_attention", + "payload": {"taskId": can_tid, "kind": "approval", "question": "Too late now."}, + }) + if att5.get("ok"): + raise AssertionError(f"request_attention must reject a terminal task: {att5!r}") + + # --- complete_project: PROJECT_COMPLETED event + idempotent retry. --- + comp_tid = "comp-task" + comp_pid = _lifecycle_setup(comp_tid) + _lifecycle_submit(comp_tid, "SUCCESS", "Comp work done.") + comp = payload("projectflow", { + "action": "complete_project", + "payload": {"projectId": comp_pid}, + }) + if not comp.get("ok"): + raise AssertionError(f"complete_project failed: {comp!r}") + note = (comp.get("project") or {}).get("projectNotification") or {} + if note.get("sent") is not True: + raise AssertionError(f"complete_project must send PROJECT_COMPLETED: {note!r}") + comp_ev = [ev for ev in matrix["events"] if f"project-{comp_pid}-" in ev["path"]] + if len(comp_ev) != 1: + raise AssertionError(f"expected one project completion event: {comp_ev!r}") + comp_body = comp_ev[0]["content"].get("body", "") + if f"PROJECT_COMPLETED: {comp_pid} - Project completed:" not in comp_body: + raise AssertionError(f"project event must carry the contract line: {comp_body!r}") + mentions = (comp_ev[0]["content"].get("m.mentions") or {}).get("user_ids", []) + if "@admin:example.test" not in mentions or "@luo:example.test" not in mentions: + raise AssertionError(f"project event must mention leader and human: {mentions!r}") + comp2 = payload("projectflow", { + "action": "complete_project", + "payload": {"projectId": comp_pid}, + }) + note2 = (comp2.get("project") or {}).get("projectNotification") or {} + if note2.get("reused") is not True or note2.get("eventId") != note.get("eventId"): + raise AssertionError(f"retried complete_project must reuse the event: {note2!r}") + if len([ev for ev in matrix["events"] if f"project-{comp_pid}-" in ev["path"]]) != 1: + raise AssertionError("retried complete_project must not send a second event") + matrix_server.shutdown() matrix_server.server_close()