Add orch and inbox CLI gap-fill tests

2026-03-24 09:40:39 +08:00
parent ea3b617ad1
commit 1bfe00b1a8
9 changed files with 4279 additions and 0 deletions
@@ -0,0 +1,208 @@
+package inbox
+
+import (
+	"strings"
+	"testing"
+)
+
+// TestInboxLifecycleCommandsRejectMissingAgentWithoutRootFallback verifies lifecycle commands require an agent when root fallback is absent.
+func TestInboxLifecycleCommandsRejectMissingAgentWithoutRootFallback(t *testing.T) {
+	t.Parallel()
+
+	cases := []struct {
+		name string
+		args []string
+	}{
+		{
+			name: "claim",
+			args: []string{"claim", "--thread", "thr_missing"},
+		},
+		{
+			name: "renew",
+			args: []string{"renew", "--thread", "thr_missing"},
+		},
+		{
+			name: "update",
+			args: []string{"update", "--thread", "thr_missing", "--status", "blocked", "--summary", "Need leader input"},
+		},
+		{
+			name: "done",
+			args: []string{"done", "--thread", "thr_missing", "--summary", "Finished"},
+		},
+		{
+			name: "fail",
+			args: []string{"fail", "--thread", "thr_missing", "--summary", "Failed"},
+		},
+		{
+			name: "cancel",
+			args: []string{"cancel", "--thread", "thr_missing"},
+		},
+	}
+
+	for _, tc := range cases {
+		tc := tc
+		t.Run(tc.name, func(t *testing.T) {
+			dbPath := initCommandTestDB(t)
+			args := append([]string{"--db", dbPath, "--json"}, tc.args...)
+
+			stdout, _, exitCode := executeInboxCommand(args...)
+			if exitCode != 30 {
+				t.Fatalf("expected invalid_input exit code 30, got %d\nstdout:\n%s", exitCode, stdout)
+			}
+			assertErrorJSON(t, stdout, "invalid_input")
+			assertInboxErrorMessageContains(t, stdout, "agent is required")
+		})
+	}
+}
+
+// TestInboxSendAndReplyRejectTerminalThreads verifies thread-append commands reject terminal threads.
+func TestInboxSendAndReplyRejectTerminalThreads(t *testing.T) {
+	t.Parallel()
+
+	t.Run("send", func(t *testing.T) {
+		dbPath := initCommandTestDB(t)
+		threadID := sendPendingThread(t, dbPath, "leader", "worker-a", "Terminal send", "Create terminal thread")
+		runInboxCommand(t, "--db", dbPath, "--json", "cancel", "--agent", "leader", "--thread", threadID, "--reason", "Stop work")
+
+		stdout, _, exitCode := executeInboxCommand(
+			"--db", dbPath,
+			"--json",
+			"send",
+			"--from", "leader",
+			"--to", "worker-a",
+			"--thread", threadID,
+			"--summary", "Retry anyway",
+		)
+		if exitCode != 30 {
+			t.Fatalf("expected invalid_state exit code 30, got %d\nstdout:\n%s", exitCode, stdout)
+		}
+		assertErrorJSON(t, stdout, "invalid_state")
+		assertInboxErrorMessageContains(t, stdout, "already terminal")
+	})
+
+	t.Run("reply", func(t *testing.T) {
+		dbPath := initCommandTestDB(t)
+		threadID := sendPendingThread(t, dbPath, "leader", "worker-a", "Terminal reply", "Create terminal thread")
+		runInboxCommand(t, "--db", dbPath, "--json", "cancel", "--agent", "leader", "--thread", threadID, "--reason", "Stop work")
+
+		stdout, _, exitCode := executeInboxCommand(
+			"--db", dbPath,
+			"--json",
+			"reply",
+			"--from", "leader",
+			"--to", "worker-a",
+			"--thread", threadID,
+			"--summary", "Reply anyway",
+		)
+		if exitCode != 30 {
+			t.Fatalf("expected invalid_state exit code 30, got %d\nstdout:\n%s", exitCode, stdout)
+		}
+		assertErrorJSON(t, stdout, "invalid_state")
+		assertInboxErrorMessageContains(t, stdout, "already terminal")
+	})
+}
+
+// TestCancelRejectsAdditionalInvalidArtifactCombinations verifies cancel covers the remaining artifact validation branches.
+func TestCancelRejectsAdditionalInvalidArtifactCombinations(t *testing.T) {
+	t.Parallel()
+
+	cases := []struct {
+		name string
+		args []string
+		want string
+	}{
+		{
+			name: "empty artifact path",
+			args: []string{"--artifact", "   "},
+			want: "artifact path cannot be empty",
+		},
+		{
+			name: "artifact kind count mismatch",
+			args: []string{"--artifact", "decision.md", "--artifact-kind", "brief", "--artifact-kind", "log"},
+			want: "artifact-kind must be specified once or once per artifact",
+		},
+		{
+			name: "artifact metadata count mismatch",
+			args: []string{"--artifact", "decision.md", "--artifact-metadata-json", `{"label":"one"}`, "--artifact-metadata-json", `{"label":"two"}`},
+			want: "artifact-metadata-json must be specified once or once per artifact",
+		},
+	}
+
+	for _, tc := range cases {
+		tc := tc
+		t.Run(tc.name, func(t *testing.T) {
+			dbPath := initCommandTestDB(t)
+			threadID := sendPendingThread(t, dbPath, "leader", "worker-a", "Cancel artifact validation", "Validate cancel artifacts")
+
+			args := []string{
+				"--db", dbPath,
+				"--json",
+				"cancel",
+				"--agent", "leader",
+				"--thread", threadID,
+			}
+			args = append(args, tc.args...)
+
+			stdout, _, exitCode := executeInboxCommand(args...)
+			if exitCode != 30 {
+				t.Fatalf("expected invalid_input exit code 30, got %d\nstdout:\n%s", exitCode, stdout)
+			}
+			assertErrorJSON(t, stdout, "invalid_input")
+			assertInboxErrorMessageContains(t, stdout, tc.want)
+		})
+	}
+}
+
+// TestInboxQueryCommandHelpContracts verifies the remaining query-style commands explain their operator contracts.
+func TestInboxQueryCommandHelpContracts(t *testing.T) {
+	t.Parallel()
+
+	cases := []struct {
+		name string
+		args []string
+		want []string
+	}{
+		{
+			name: "send",
+			args: []string{"send", "--help"},
+			want: []string{"low-level thread creation primitive", "Constraints:", "body-file ./brief.md"},
+		},
+		{
+			name: "reply",
+			args: []string{"reply", "--help"},
+			want: []string{"append a directed answer, control note, or follow-up message", "Constraints:", "--kind control"},
+		},
+		{
+			name: "fetch",
+			args: []string{"fetch", "--help"},
+			want: []string{"fetch -> claim -> show -> update/done/fail", "Constraints:", "--status pending,blocked --unread --limit 5"},
+		},
+		{
+			name: "show",
+			args: []string{"show", "--help"},
+			want: []string{"exact message sequence inside one thread", "Constraints:", "--mark-read"},
+		},
+		{
+			name: "watch",
+			args: []string{"watch", "--help"},
+			want: []string{"watch is broader than wait-reply", "Constraints:", "--after-event 100 --timeout-seconds 300"},
+		},
+	}
+
+	for _, tc := range cases {
+		tc := tc
+		t.Run(tc.name, func(t *testing.T) {
+			stdout, stderr, exitCode := executeInboxCommand(tc.args...)
+			if exitCode != 0 {
+				t.Fatalf("expected help exit 0, got %d\nstderr:\n%s\nstdout:\n%s", exitCode, stderr, stdout)
+			}
+
+			combined := stdout + stderr
+			for _, want := range tc.want {
+				if !strings.Contains(combined, want) {
+					t.Fatalf("expected %s help to contain %q, got:\n%s", tc.name, want, combined)
+				}
+			}
+		})
+	}
+}
@@ -0,0 +1,629 @@
+package inbox
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+)
+
+func releaseLeaseForLifecycleTest(t *testing.T, dbPath, threadID string) {
+	t.Helper()
+
+	sqlDB, err := openInboxDB(context.Background(), dbPath)
+	if err != nil {
+		t.Fatalf("open db: %v", err)
+	}
+	defer sqlDB.Close()
+
+	if _, err := sqlDB.ExecContext(context.Background(), `UPDATE leases SET released_at = expires_at WHERE thread_id = ?`, threadID); err != nil {
+		t.Fatalf("release lease: %v", err)
+	}
+}
+
+func assertInboxErrorMessageContains(t *testing.T, raw string, want string) {
+	t.Helper()
+
+	var payload map[string]any
+	mustDecodeJSON(t, raw, &payload)
+	errorValue, ok := payload["error"].(map[string]any)
+	if !ok {
+		t.Fatalf("expected error object, got %#v", payload["error"])
+	}
+	message, _ := errorValue["message"].(string)
+	if !strings.Contains(message, want) {
+		t.Fatalf("expected error message to contain %q, got %q", want, message)
+	}
+}
+
+// TestClaimRejectsNonPendingThread verifies claim rejects threads that are no longer pending.
+func TestClaimRejectsNonPendingThread(t *testing.T) {
+	t.Parallel()
+
+	dbPath := initCommandTestDB(t)
+	threadID := sendPendingThread(t, dbPath, "leader", "worker-a", "Claimed already", "Trigger invalid state")
+
+	runInboxCommand(
+		t,
+		"--db", dbPath,
+		"--json",
+		"claim",
+		"--agent", "worker-a",
+		"--thread", threadID,
+	)
+	releaseLeaseForLifecycleTest(t, dbPath, threadID)
+
+	stdout, _, exitCode := executeInboxCommand(
+		"--db", dbPath,
+		"--json",
+		"claim",
+		"--agent", "worker-a",
+		"--thread", threadID,
+	)
+	if exitCode != 30 {
+		t.Fatalf("expected exit code 30, got %d with output %s", exitCode, stdout)
+	}
+	assertErrorJSON(t, stdout, "invalid_state")
+	assertInboxErrorMessageContains(t, stdout, "is not pending")
+}
+
+// TestInboxLifecycleCommandsRespectRootAgentAndPlainTextOutput verifies lifecycle commands accept the root agent flag and render text output.
+func TestInboxLifecycleCommandsRespectRootAgentAndPlainTextOutput(t *testing.T) {
+	t.Parallel()
+
+	t.Run("claim", func(t *testing.T) {
+		dbPath := initCommandTestDB(t)
+		threadID := sendPendingThread(t, dbPath, "leader", "worker-a", "Claim root agent", "Use root agent")
+
+		stdout, stderr, exitCode := executeInboxCommand(
+			"--db", dbPath,
+			"--agent", "worker-a",
+			"claim",
+			"--thread", threadID,
+		)
+		if exitCode != 0 {
+			t.Fatalf("expected exit 0, got %d\nstderr:\n%s\nstdout:\n%s", exitCode, stderr, stdout)
+		}
+		if !strings.Contains(stdout, "claimed thread "+threadID) {
+			t.Fatalf("expected claim output to mention thread %s, got %q", threadID, stdout)
+		}
+	})
+
+	t.Run("renew", func(t *testing.T) {
+		dbPath := initCommandTestDB(t)
+		threadID := sendPendingThread(t, dbPath, "leader", "worker-b", "Renew root agent", "Use root agent")
+		runInboxCommand(t, "--db", dbPath, "--json", "claim", "--agent", "worker-b", "--thread", threadID)
+
+		stdout, stderr, exitCode := executeInboxCommand(
+			"--db", dbPath,
+			"--agent", "worker-b",
+			"renew",
+			"--thread", threadID,
+		)
+		if exitCode != 0 {
+			t.Fatalf("expected exit 0, got %d\nstderr:\n%s\nstdout:\n%s", exitCode, stderr, stdout)
+		}
+		if !strings.Contains(stdout, "renewed lease on thread "+threadID) {
+			t.Fatalf("expected renew output to mention thread %s, got %q", threadID, stdout)
+		}
+	})
+
+	t.Run("update", func(t *testing.T) {
+		dbPath := filepath.Join(t.TempDir(), "coord.db")
+		threadID := seedClaimedThreadForInboxTests(t, dbPath, "leader", "worker-c", "worker-c")
+
+		stdout, stderr, exitCode := executeInboxCommand(
+			"--db", dbPath,
+			"--agent", "worker-c",
+			"update",
+			"--thread", threadID,
+			"--status", "in_progress",
+			"--summary", "Started from root agent",
+		)
+		if exitCode != 0 {
+			t.Fatalf("expected exit 0, got %d\nstderr:\n%s\nstdout:\n%s", exitCode, stderr, stdout)
+		}
+		if !strings.Contains(stdout, "updated thread "+threadID+" to in_progress") {
+			t.Fatalf("expected update output to mention thread %s, got %q", threadID, stdout)
+		}
+	})
+
+	t.Run("done", func(t *testing.T) {
+		dbPath := filepath.Join(t.TempDir(), "coord.db")
+		threadID := seedClaimedThreadForInboxTests(t, dbPath, "leader", "worker-d", "worker-d")
+
+		stdout, stderr, exitCode := executeInboxCommand(
+			"--db", dbPath,
+			"--agent", "worker-d",
+			"done",
+			"--thread", threadID,
+			"--summary", "Finished via root agent",
+		)
+		if exitCode != 0 {
+			t.Fatalf("expected exit 0, got %d\nstderr:\n%s\nstdout:\n%s", exitCode, stderr, stdout)
+		}
+		if !strings.Contains(stdout, "done thread "+threadID) {
+			t.Fatalf("expected done output to mention thread %s, got %q", threadID, stdout)
+		}
+	})
+
+	t.Run("fail", func(t *testing.T) {
+		dbPath := filepath.Join(t.TempDir(), "coord.db")
+		threadID := seedClaimedThreadForInboxTests(t, dbPath, "leader", "worker-e", "worker-e")
+
+		stdout, stderr, exitCode := executeInboxCommand(
+			"--db", dbPath,
+			"--agent", "worker-e",
+			"fail",
+			"--thread", threadID,
+			"--summary", "Failed via root agent",
+		)
+		if exitCode != 0 {
+			t.Fatalf("expected exit 0, got %d\nstderr:\n%s\nstdout:\n%s", exitCode, stderr, stdout)
+		}
+		if !strings.Contains(stdout, "fail thread "+threadID) {
+			t.Fatalf("expected fail output to mention thread %s, got %q", threadID, stdout)
+		}
+	})
+
+	t.Run("cancel", func(t *testing.T) {
+		dbPath := initCommandTestDB(t)
+		threadID := sendPendingThread(t, dbPath, "leader", "worker-f", "Cancel root agent", "Use root agent")
+
+		stdout, stderr, exitCode := executeInboxCommand(
+			"--db", dbPath,
+			"--agent", "leader",
+			"cancel",
+			"--thread", threadID,
+			"--reason", "Task superseded",
+		)
+		if exitCode != 0 {
+			t.Fatalf("expected exit 0, got %d\nstderr:\n%s\nstdout:\n%s", exitCode, stderr, stdout)
+		}
+		if !strings.Contains(stdout, "cancelled thread "+threadID) {
+			t.Fatalf("expected cancel output to mention thread %s, got %q", threadID, stdout)
+		}
+	})
+}
+
+// TestInboxLifecycleCommandHelpContracts verifies claim, renew, done, fail, and cancel help text explains their contracts.
+func TestInboxLifecycleCommandHelpContracts(t *testing.T) {
+	t.Parallel()
+
+	cases := []struct {
+		name string
+		args []string
+		want []string
+	}{
+		{
+			name: "claim",
+			args: []string{"claim", "--help"},
+			want: []string{"Only one active lease may exist per thread at a time.", "Constraints:", "--lease-seconds 1800"},
+		},
+		{
+			name: "renew",
+			args: []string{"renew", "--help"},
+			want: []string{"renew only applies when the caller already owns the active lease.", "Constraints:", "--lease-seconds 1800"},
+		},
+		{
+			name: "done",
+			args: []string{"done", "--help"},
+			want: []string{"done is a terminal operation", "Constraints:", "--summary \"Implemented retry policy\""},
+		},
+		{
+			name: "fail",
+			args: []string{"fail", "--help"},
+			want: []string{"fail is a terminal operation", "Constraints:", "retry, reassignment, or cancellation"},
+		},
+		{
+			name: "cancel",
+			args: []string{"cancel", "--help"},
+			want: []string{"leader or operator action", "Constraints:", "--reason \"Task superseded by a new plan.\""},
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			stdout, stderr, exitCode := executeInboxCommand(tc.args...)
+			if exitCode != 0 {
+				t.Fatalf("expected help exit 0, got %d\nstderr:\n%s\nstdout:\n%s", exitCode, stderr, stdout)
+			}
+
+			combined := stdout + stderr
+			for _, want := range tc.want {
+				if !strings.Contains(combined, want) {
+					t.Fatalf("expected help for %s to contain %q, got:\n%s", tc.name, want, combined)
+				}
+			}
+		})
+	}
+}
+
+// TestRenewRejectsMissingAndTerminalThread verifies renew reports not found and terminal-thread errors.
+func TestRenewRejectsMissingAndTerminalThread(t *testing.T) {
+	t.Parallel()
+
+	t.Run("missing thread", func(t *testing.T) {
+		dbPath := initCommandTestDB(t)
+
+		stdout, _, exitCode := executeInboxCommand(
+			"--db", dbPath,
+			"--json",
+			"renew",
+			"--agent", "worker-a",
+			"--thread", "thr_missing",
+		)
+		if exitCode != 40 {
+			t.Fatalf("expected exit code 40, got %d with output %s", exitCode, stdout)
+		}
+		assertErrorJSON(t, stdout, "not_found")
+	})
+
+	t.Run("terminal thread", func(t *testing.T) {
+		dbPath := filepath.Join(t.TempDir(), "coord.db")
+		threadID := seedClaimedThreadForInboxTests(t, dbPath, "leader", "worker-a", "worker-a")
+		runInboxCommand(t, "--db", dbPath, "--json", "done", "--agent", "worker-a", "--thread", threadID, "--summary", "Finished")
+
+		stdout, _, exitCode := executeInboxCommand(
+			"--db", dbPath,
+			"--json",
+			"renew",
+			"--agent", "worker-a",
+			"--thread", threadID,
+		)
+		if exitCode != 30 {
+			t.Fatalf("expected exit code 30, got %d with output %s", exitCode, stdout)
+		}
+		assertErrorJSON(t, stdout, "invalid_state")
+		assertInboxErrorMessageContains(t, stdout, "already terminal")
+	})
+}
+
+// TestUpdateRejectsMissingAndTerminalThread verifies update rejects unknown or terminal threads.
+func TestUpdateRejectsMissingAndTerminalThread(t *testing.T) {
+	t.Parallel()
+
+	t.Run("missing thread", func(t *testing.T) {
+		dbPath := initCommandTestDB(t)
+
+		stdout, _, exitCode := executeInboxCommand(
+			"--db", dbPath,
+			"--json",
+			"update",
+			"--agent", "worker-a",
+			"--thread", "thr_missing",
+			"--status", "in_progress",
+			"--summary", "Started",
+		)
+		if exitCode != 40 {
+			t.Fatalf("expected exit code 40, got %d with output %s", exitCode, stdout)
+		}
+		assertErrorJSON(t, stdout, "not_found")
+	})
+
+	t.Run("terminal thread", func(t *testing.T) {
+		dbPath := filepath.Join(t.TempDir(), "coord.db")
+		threadID := seedClaimedThreadForInboxTests(t, dbPath, "leader", "worker-a", "worker-a")
+		runInboxCommand(t, "--db", dbPath, "--json", "done", "--agent", "worker-a", "--thread", threadID, "--summary", "Finished")
+
+		stdout, _, exitCode := executeInboxCommand(
+			"--db", dbPath,
+			"--json",
+			"update",
+			"--agent", "worker-a",
+			"--thread", threadID,
+			"--status", "blocked",
+			"--summary", "Need answer",
+		)
+		if exitCode != 30 {
+			t.Fatalf("expected exit code 30, got %d with output %s", exitCode, stdout)
+		}
+		assertErrorJSON(t, stdout, "invalid_state")
+		assertInboxErrorMessageContains(t, stdout, "already terminal")
+	})
+}
+
+// TestUpdateRejectsInvalidBodyFileInputs verifies update rejects mutually exclusive or unreadable body-file inputs.
+func TestUpdateRejectsInvalidBodyFileInputs(t *testing.T) {
+	t.Parallel()
+
+	dbPath := filepath.Join(t.TempDir(), "coord.db")
+	threadID := seedClaimedThreadForInboxTests(t, dbPath, "leader", "worker-a", "worker-a")
+
+	t.Run("body and body-file", func(t *testing.T) {
+		tempDir := t.TempDir()
+		bodyPath := filepath.Join(tempDir, "body.txt")
+		if err := os.WriteFile(bodyPath, []byte("from file"), 0o644); err != nil {
+			t.Fatalf("write body file: %v", err)
+		}
+
+		stdout, _, exitCode := executeInboxCommand(
+			"--db", dbPath,
+			"--json",
+			"update",
+			"--agent", "worker-a",
+			"--thread", threadID,
+			"--status", "in_progress",
+			"--summary", "Started",
+			"--body", "inline",
+			"--body-file", bodyPath,
+		)
+		if exitCode != 30 {
+			t.Fatalf("expected exit code 30, got %d with output %s", exitCode, stdout)
+		}
+		assertErrorJSON(t, stdout, "invalid_input")
+		assertInboxErrorMessageContains(t, stdout, "mutually exclusive")
+	})
+
+	t.Run("unreadable body-file", func(t *testing.T) {
+		stdout, _, exitCode := executeInboxCommand(
+			"--db", dbPath,
+			"--json",
+			"update",
+			"--agent", "worker-a",
+			"--thread", threadID,
+			"--status", "in_progress",
+			"--summary", "Started",
+			"--body-file", filepath.Join(t.TempDir(), "missing.txt"),
+		)
+		if exitCode != 30 {
+			t.Fatalf("expected exit code 30, got %d with output %s", exitCode, stdout)
+		}
+		assertErrorJSON(t, stdout, "invalid_input")
+		assertInboxErrorMessageContains(t, stdout, "failed to read body-file")
+	})
+}
+
+// TestUpdateRejectsInvalidArtifactFlagCombinations verifies update validates artifact flag combinations before command execution.
+func TestUpdateRejectsInvalidArtifactFlagCombinations(t *testing.T) {
+	t.Parallel()
+
+	dbPath := filepath.Join(t.TempDir(), "coord.db")
+	threadID := seedClaimedThreadForInboxTests(t, dbPath, "leader", "worker-a", "worker-a")
+
+	cases := []struct {
+		name string
+		args []string
+		want string
+	}{
+		{
+			name: "kind without artifact",
+			args: []string{"--artifact-kind", "log"},
+			want: "require at least one artifact path",
+		},
+		{
+			name: "empty artifact path",
+			args: []string{"--artifact", "   "},
+			want: "artifact path cannot be empty",
+		},
+		{
+			name: "artifact-kind count mismatch",
+			args: []string{"--artifact", "a.log", "--artifact", "b.log", "--artifact-kind", "log", "--artifact-kind", "brief", "--artifact-kind", "extra"},
+			want: "artifact-kind must be specified once or once per artifact",
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			args := []string{
+				"--db", dbPath,
+				"--json",
+				"update",
+				"--agent", "worker-a",
+				"--thread", threadID,
+				"--status", "in_progress",
+				"--summary", "Started",
+			}
+			args = append(args, tc.args...)
+
+			stdout, _, exitCode := executeInboxCommand(args...)
+			if exitCode != 30 {
+				t.Fatalf("expected exit code 30, got %d with output %s", exitCode, stdout)
+			}
+			assertErrorJSON(t, stdout, "invalid_input")
+			assertInboxErrorMessageContains(t, stdout, tc.want)
+		})
+	}
+}
+
+// TestDoneAndFailRejectMissingThread verifies terminal commands reject unknown threads.
+func TestDoneAndFailRejectMissingThread(t *testing.T) {
+	t.Parallel()
+
+	for _, command := range []string{"done", "fail"} {
+		t.Run(command, func(t *testing.T) {
+			dbPath := initCommandTestDB(t)
+
+			stdout, _, exitCode := executeInboxCommand(
+				"--db", dbPath,
+				"--json",
+				command,
+				"--agent", "worker-a",
+				"--thread", "thr_missing",
+				"--summary", "Missing thread",
+			)
+			if exitCode != 40 {
+				t.Fatalf("expected exit code 40, got %d with output %s", exitCode, stdout)
+			}
+			assertErrorJSON(t, stdout, "not_found")
+		})
+	}
+}
+
+// TestDoneAndFailRejectInvalidBodyFileInputs verifies terminal commands reject bad body-file inputs.
+func TestDoneAndFailRejectInvalidBodyFileInputs(t *testing.T) {
+	t.Parallel()
+
+	cases := []struct {
+		command string
+		agent   string
+	}{
+		{command: "done", agent: "worker-a"},
+		{command: "fail", agent: "worker-b"},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.command, func(t *testing.T) {
+			dbPath := filepath.Join(t.TempDir(), "coord.db")
+			threadID := seedClaimedThreadForInboxTests(t, dbPath, "leader", tc.agent, tc.agent)
+			bodyPath := filepath.Join(t.TempDir(), "body.txt")
+			if err := os.WriteFile(bodyPath, []byte("from file"), 0o644); err != nil {
+				t.Fatalf("write body file: %v", err)
+			}
+
+			stdout, _, exitCode := executeInboxCommand(
+				"--db", dbPath,
+				"--json",
+				tc.command,
+				"--agent", tc.agent,
+				"--thread", threadID,
+				"--summary", "Finished",
+				"--body", "inline",
+				"--body-file", bodyPath,
+			)
+			if exitCode != 30 {
+				t.Fatalf("expected exit code 30, got %d with output %s", exitCode, stdout)
+			}
+			assertErrorJSON(t, stdout, "invalid_input")
+			assertInboxErrorMessageContains(t, stdout, "mutually exclusive")
+
+			stdout, _, exitCode = executeInboxCommand(
+				"--db", dbPath,
+				"--json",
+				tc.command,
+				"--agent", tc.agent,
+				"--thread", threadID,
+				"--summary", "Finished",
+				"--body-file", filepath.Join(t.TempDir(), "missing.txt"),
+			)
+			if exitCode != 30 {
+				t.Fatalf("expected exit code 30, got %d with output %s", exitCode, stdout)
+			}
+			assertErrorJSON(t, stdout, "invalid_input")
+			assertInboxErrorMessageContains(t, stdout, "failed to read body-file")
+		})
+	}
+}
+
+// TestDoneAndFailRejectInvalidArtifactFlagCombinations verifies terminal commands validate artifact flag combinations.
+func TestDoneAndFailRejectInvalidArtifactFlagCombinations(t *testing.T) {
+	t.Parallel()
+
+	cases := []struct {
+		command string
+		agent   string
+		args    []string
+		want    string
+	}{
+		{
+			command: "done",
+			agent:   "worker-a",
+			args:    []string{"--artifact-kind", "brief"},
+			want:    "require at least one artifact path",
+		},
+		{
+			command: "fail",
+			agent:   "worker-b",
+			args:    []string{"--artifact", "a.log", "--artifact", "b.log", "--artifact-metadata-json", `{}`, "--artifact-metadata-json", `{}`, "--artifact-metadata-json", `{}`},
+			want:    "artifact-metadata-json must be specified once or once per artifact",
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.command, func(t *testing.T) {
+			dbPath := filepath.Join(t.TempDir(), "coord.db")
+			threadID := seedClaimedThreadForInboxTests(t, dbPath, "leader", tc.agent, tc.agent)
+
+			args := []string{
+				"--db", dbPath,
+				"--json",
+				tc.command,
+				"--agent", tc.agent,
+				"--thread", threadID,
+				"--summary", "Finished",
+			}
+			args = append(args, tc.args...)
+
+			stdout, _, exitCode := executeInboxCommand(args...)
+			if exitCode != 30 {
+				t.Fatalf("expected exit code 30, got %d with output %s", exitCode, stdout)
+			}
+			assertErrorJSON(t, stdout, "invalid_input")
+			assertInboxErrorMessageContains(t, stdout, tc.want)
+		})
+	}
+}
+
+// TestCancelRejectsTerminalThread verifies cancel rejects threads that are already terminal.
+func TestCancelRejectsTerminalThread(t *testing.T) {
+	t.Parallel()
+
+	dbPath := filepath.Join(t.TempDir(), "coord.db")
+	threadID := seedClaimedThreadForInboxTests(t, dbPath, "leader", "worker-a", "worker-a")
+	runInboxCommand(t, "--db", dbPath, "--json", "done", "--agent", "worker-a", "--thread", threadID, "--summary", "Finished")
+
+	stdout, _, exitCode := executeInboxCommand(
+		"--db", dbPath,
+		"--json",
+		"cancel",
+		"--agent", "leader",
+		"--thread", threadID,
+	)
+	if exitCode != 30 {
+		t.Fatalf("expected exit code 30, got %d with output %s", exitCode, stdout)
+	}
+	assertErrorJSON(t, stdout, "invalid_state")
+	assertInboxErrorMessageContains(t, stdout, "already terminal")
+}
+
+// TestCancelUsesDefaultReasonWhenOmitted verifies cancel falls back to the default summary when no reason is supplied.
+func TestCancelUsesDefaultReasonWhenOmitted(t *testing.T) {
+	t.Parallel()
+
+	dbPath := initCommandTestDB(t)
+	threadID := sendPendingThread(t, dbPath, "leader", "worker-a", "Default cancel reason", "Check default summary")
+
+	runInboxCommand(
+		t,
+		"--db", dbPath,
+		"--json",
+		"cancel",
+		"--agent", "leader",
+		"--thread", threadID,
+	)
+
+	showOut := runInboxCommand(t, "--db", dbPath, "--json", "show", "--thread", threadID)
+	var showResp map[string]any
+	mustDecodeJSON(t, showOut, &showResp)
+
+	lastMessage := lastThreadMessageFromShow(t, showResp)
+	if got := lastMessage["summary"]; got != "thread cancelled" {
+		t.Fatalf("expected default cancel summary, got %#v", got)
+	}
+	if got := lastMessage["body"]; got != "" {
+		t.Fatalf("expected empty cancel body when reason omitted, got %#v", got)
+	}
+}
+
+// TestCancelRejectsInvalidArtifactFlagCombinations verifies cancel validates artifact flag combinations.
+func TestCancelRejectsInvalidArtifactFlagCombinations(t *testing.T) {
+	t.Parallel()
+
+	dbPath := initCommandTestDB(t)
+	threadID := sendPendingThread(t, dbPath, "leader", "worker-a", "Cancel invalid artifact", "Check artifact validation")
+
+	stdout, _, exitCode := executeInboxCommand(
+		"--db", dbPath,
+		"--json",
+		"cancel",
+		"--agent", "leader",
+		"--thread", threadID,
+		"--artifact-metadata-json", `{}`,
+	)
+	if exitCode != 30 {
+		t.Fatalf("expected exit code 30, got %d with output %s", exitCode, stdout)
+	}
+	assertErrorJSON(t, stdout, "invalid_input")
+	assertInboxErrorMessageContains(t, stdout, "require at least one artifact path")
+}
@@ -0,0 +1,597 @@
+package inbox
+
+import (
+	"os"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"testing"
+	"time"
+)
+
+// TestSendRejectsMissingFromAgent verifies send requires a sender when no root agent is set.
+func TestSendRejectsMissingFromAgent(t *testing.T) {
+	t.Parallel()
+
+	dbPath := initCommandTestDB(t)
+
+	stdout, _, exitCode := executeInboxCommand(
+		"--db", dbPath,
+		"--json",
+		"send",
+		"--to", "worker-a",
+		"--subject", "Investigate flaky test",
+	)
+	if exitCode != 30 {
+		t.Fatalf("expected invalid_input exit code 30, got %d\nstdout:\n%s", exitCode, stdout)
+	}
+	assertErrorJSON(t, stdout, "invalid_input")
+	assertQueryWaitErrorMessageContains(t, stdout, "from agent is required")
+}
+
+// TestSendRejectsMissingSubjectWhenCreatingThread verifies send requires subject for a new thread.
+func TestSendRejectsMissingSubjectWhenCreatingThread(t *testing.T) {
+	t.Parallel()
+
+	dbPath := initCommandTestDB(t)
+
+	stdout, _, exitCode := executeInboxCommand(
+		"--db", dbPath,
+		"--json",
+		"send",
+		"--from", "leader",
+		"--to", "worker-a",
+		"--summary", "Missing subject",
+	)
+	if exitCode != 30 {
+		t.Fatalf("expected invalid_input exit code 30, got %d\nstdout:\n%s", exitCode, stdout)
+	}
+	assertErrorJSON(t, stdout, "invalid_input")
+	assertQueryWaitErrorMessageContains(t, stdout, "subject is required when creating a new thread")
+}
+
+// TestSendRejectsBodyAndBodyFileTogether verifies send rejects mutually exclusive body inputs.
+func TestSendRejectsBodyAndBodyFileTogether(t *testing.T) {
+	t.Parallel()
+
+	tempDir := t.TempDir()
+	dbPath := filepath.Join(tempDir, "coord.db")
+	bodyPath := filepath.Join(tempDir, "brief.md")
+	if err := os.WriteFile(bodyPath, []byte("brief"), 0o644); err != nil {
+		t.Fatalf("write body file: %v", err)
+	}
+	runInboxCommand(t, "--db", dbPath, "--json", "init")
+
+	stdout, _, exitCode := executeInboxCommand(
+		"--db", dbPath,
+		"--json",
+		"send",
+		"--from", "leader",
+		"--to", "worker-a",
+		"--subject", "Investigate flaky test",
+		"--body", "inline",
+		"--body-file", bodyPath,
+	)
+	if exitCode != 30 {
+		t.Fatalf("expected invalid_input exit code 30, got %d\nstdout:\n%s", exitCode, stdout)
+	}
+	assertErrorJSON(t, stdout, "invalid_input")
+	assertQueryWaitErrorMessageContains(t, stdout, "body and body-file are mutually exclusive")
+}
+
+// TestSendRejectsUnreadableBodyFile verifies send reports unreadable body files.
+func TestSendRejectsUnreadableBodyFile(t *testing.T) {
+	t.Parallel()
+
+	dbPath := initCommandTestDB(t)
+
+	stdout, _, exitCode := executeInboxCommand(
+		"--db", dbPath,
+		"--json",
+		"send",
+		"--from", "leader",
+		"--to", "worker-a",
+		"--subject", "Investigate flaky test",
+		"--body-file", filepath.Join(t.TempDir(), "missing.md"),
+	)
+	if exitCode != 30 {
+		t.Fatalf("expected invalid_input exit code 30, got %d\nstdout:\n%s", exitCode, stdout)
+	}
+	assertErrorJSON(t, stdout, "invalid_input")
+	assertQueryWaitErrorMessageContains(t, stdout, "failed to read body-file")
+}
+
+// TestSendRejectsArtifactKindCountMismatch verifies send validates artifact flag cardinality.
+func TestSendRejectsArtifactKindCountMismatch(t *testing.T) {
+	t.Parallel()
+
+	dbPath := initCommandTestDB(t)
+
+	stdout, _, exitCode := executeInboxCommand(
+		"--db", dbPath,
+		"--json",
+		"send",
+		"--from", "leader",
+		"--to", "worker-a",
+		"--subject", "Investigate flaky test",
+		"--artifact", "/tmp/task.md",
+		"--artifact-kind", "brief",
+		"--artifact-kind", "log",
+	)
+	if exitCode != 30 {
+		t.Fatalf("expected invalid_input exit code 30, got %d\nstdout:\n%s", exitCode, stdout)
+	}
+	assertErrorJSON(t, stdout, "invalid_input")
+	assertQueryWaitErrorMessageContains(t, stdout, "artifact-kind must be specified once or once per artifact")
+}
+
+// TestSendUsesRootAgentAndPlainTextOutput verifies send falls back to the root agent and renders plain text output.
+func TestSendUsesRootAgentAndPlainTextOutput(t *testing.T) {
+	t.Parallel()
+
+	dbPath := initCommandTestDB(t)
+
+	stdout := runInboxCommand(
+		t,
+		"--db", dbPath,
+		"--agent", "leader",
+		"send",
+		"--to", "worker-a",
+		"--subject", "Investigate flaky test",
+		"--summary", "Check latest failures",
+	)
+	threadID := parsePlainThreadID(t, stdout, "created thread ")
+
+	showOut := runInboxCommand(
+		t,
+		"--db", dbPath,
+		"--json",
+		"show",
+		"--thread", threadID,
+	)
+
+	var showResp map[string]any
+	mustDecodeJSON(t, showOut, &showResp)
+	if got := nestedString(t, showResp, "data", "thread", "created_by"); got != "leader" {
+		t.Fatalf("expected created_by leader, got %q", got)
+	}
+}
+
+// TestReplyRejectsMissingFromAgent verifies reply requires a sender when no root agent is set.
+func TestReplyRejectsMissingFromAgent(t *testing.T) {
+	t.Parallel()
+
+	dbPath := filepath.Join(t.TempDir(), "coord.db")
+	threadID := seedThreadForInboxTests(t, dbPath, "leader", "worker-a")
+
+	stdout, _, exitCode := executeInboxCommand(
+		"--db", dbPath,
+		"--json",
+		"reply",
+		"--to", "worker-a",
+		"--thread", threadID,
+		"--summary", "Use stdout",
+	)
+	if exitCode != 30 {
+		t.Fatalf("expected invalid_input exit code 30, got %d\nstdout:\n%s", exitCode, stdout)
+	}
+	assertErrorJSON(t, stdout, "invalid_input")
+	assertQueryWaitErrorMessageContains(t, stdout, "from agent is required")
+}
+
+// TestReplyRejectsBodyAndBodyFileTogether verifies reply rejects mutually exclusive body inputs.
+func TestReplyRejectsBodyAndBodyFileTogether(t *testing.T) {
+	t.Parallel()
+
+	tempDir := t.TempDir()
+	dbPath := filepath.Join(tempDir, "coord.db")
+	bodyPath := filepath.Join(tempDir, "reply.md")
+	if err := os.WriteFile(bodyPath, []byte("reply"), 0o644); err != nil {
+		t.Fatalf("write body file: %v", err)
+	}
+	threadID := seedThreadForInboxTests(t, dbPath, "leader", "worker-a")
+
+	stdout, _, exitCode := executeInboxCommand(
+		"--db", dbPath,
+		"--json",
+		"reply",
+		"--from", "leader",
+		"--to", "worker-a",
+		"--thread", threadID,
+		"--summary", "Use stdout",
+		"--body", "inline",
+		"--body-file", bodyPath,
+	)
+	if exitCode != 30 {
+		t.Fatalf("expected invalid_input exit code 30, got %d\nstdout:\n%s", exitCode, stdout)
+	}
+	assertErrorJSON(t, stdout, "invalid_input")
+	assertQueryWaitErrorMessageContains(t, stdout, "body and body-file are mutually exclusive")
+}
+
+// TestReplyRejectsUnreadableBodyFile verifies reply reports unreadable body files.
+func TestReplyRejectsUnreadableBodyFile(t *testing.T) {
+	t.Parallel()
+
+	dbPath := filepath.Join(t.TempDir(), "coord.db")
+	threadID := seedThreadForInboxTests(t, dbPath, "leader", "worker-a")
+
+	stdout, _, exitCode := executeInboxCommand(
+		"--db", dbPath,
+		"--json",
+		"reply",
+		"--from", "leader",
+		"--to", "worker-a",
+		"--thread", threadID,
+		"--summary", "Use stdout",
+		"--body-file", filepath.Join(t.TempDir(), "missing.md"),
+	)
+	if exitCode != 30 {
+		t.Fatalf("expected invalid_input exit code 30, got %d\nstdout:\n%s", exitCode, stdout)
+	}
+	assertErrorJSON(t, stdout, "invalid_input")
+	assertQueryWaitErrorMessageContains(t, stdout, "failed to read body-file")
+}
+
+// TestReplyRejectsArtifactMetadataCountMismatch verifies reply validates artifact metadata cardinality.
+func TestReplyRejectsArtifactMetadataCountMismatch(t *testing.T) {
+	t.Parallel()
+
+	dbPath := filepath.Join(t.TempDir(), "coord.db")
+	threadID := seedThreadForInboxTests(t, dbPath, "leader", "worker-a")
+
+	stdout, _, exitCode := executeInboxCommand(
+		"--db", dbPath,
+		"--json",
+		"reply",
+		"--from", "leader",
+		"--to", "worker-a",
+		"--thread", threadID,
+		"--summary", "Use stdout",
+		"--artifact", "/tmp/decision.md",
+		"--artifact-metadata-json", `{"label":"a"}`,
+		"--artifact-metadata-json", `{"label":"b"}`,
+	)
+	if exitCode != 30 {
+		t.Fatalf("expected invalid_input exit code 30, got %d\nstdout:\n%s", exitCode, stdout)
+	}
+	assertErrorJSON(t, stdout, "invalid_input")
+	assertQueryWaitErrorMessageContains(t, stdout, "artifact-metadata-json must be specified once or once per artifact")
+}
+
+// TestReplyUsesRootAgentAndPlainTextOutput verifies reply falls back to the root agent and renders plain text output.
+func TestReplyUsesRootAgentAndPlainTextOutput(t *testing.T) {
+	t.Parallel()
+
+	dbPath := filepath.Join(t.TempDir(), "coord.db")
+	threadID := seedThreadForInboxTests(t, dbPath, "leader", "worker-a")
+
+	stdout := runInboxCommand(
+		t,
+		"--db", dbPath,
+		"--agent", "leader",
+		"reply",
+		"--to", "worker-a",
+		"--thread", threadID,
+		"--summary", "Use stdout",
+	)
+	if got := strings.TrimSpace(stdout); got != "replied on thread "+threadID {
+		t.Fatalf("expected plain text reply output for %q, got %q", threadID, got)
+	}
+
+	showOut := runInboxCommand(t, "--db", dbPath, "--json", "show", "--thread", threadID)
+	var showResp map[string]any
+	mustDecodeJSON(t, showOut, &showResp)
+	lastMessage := lastThreadMessageFromShow(t, showResp)
+	if got, _ := lastMessage["from_agent"].(string); got != "leader" {
+		t.Fatalf("expected last from_agent leader, got %#v", lastMessage["from_agent"])
+	}
+}
+
+// TestFetchDefaultsToPendingWithRootAgentAndPlainText verifies fetch uses pending by default and respects the root agent.
+func TestFetchDefaultsToPendingWithRootAgentAndPlainText(t *testing.T) {
+	t.Parallel()
+
+	dbPath := filepath.Join(t.TempDir(), "coord.db")
+	runInboxCommand(t, "--db", dbPath, "--json", "init")
+
+	pendingThreadID := sendPendingThread(t, dbPath, "leader", "worker-a", "Pending work", "Check latest failures")
+	blockedThreadID := sendPendingThread(t, dbPath, "leader", "worker-a", "Blocked work", "Need product decision")
+	runInboxCommand(t, "--db", dbPath, "--json", "claim", "--agent", "worker-a", "--thread", blockedThreadID)
+	runInboxCommand(
+		t,
+		"--db", dbPath,
+		"--json",
+		"update",
+		"--agent", "worker-a",
+		"--thread", blockedThreadID,
+		"--status", "blocked",
+		"--summary", "Need product decision",
+	)
+
+	stdout := runInboxCommand(t, "--db", dbPath, "--agent", "worker-a", "fetch")
+	lines := splitNonEmptyLines(stdout)
+	if len(lines) != 1 {
+		t.Fatalf("expected one plain text fetch row, got %#v", lines)
+	}
+	if !strings.Contains(lines[0], pendingThreadID+"\tpending\tPending work") {
+		t.Fatalf("expected pending thread %q in fetch output, got %q", pendingThreadID, lines[0])
+	}
+	if strings.Contains(lines[0], blockedThreadID) {
+		t.Fatalf("did not expect blocked thread %q in default fetch output %q", blockedThreadID, lines[0])
+	}
+}
+
+// TestListReturnsNoMatchingWork verifies list returns no_matching_work when filters match nothing.
+func TestListReturnsNoMatchingWork(t *testing.T) {
+	t.Parallel()
+
+	dbPath := filepath.Join(t.TempDir(), "coord.db")
+	runInboxCommand(t, "--db", dbPath, "--json", "init")
+	sendPendingThread(t, dbPath, "leader", "worker-a", "Pending work", "Check latest failures")
+
+	stdout, _, exitCode := executeInboxCommand(
+		"--db", dbPath,
+		"--json",
+		"list",
+		"--assigned-to", "worker-z",
+	)
+	if exitCode != 10 {
+		t.Fatalf("expected no_matching_work exit code 10, got %d\nstdout:\n%s", exitCode, stdout)
+	}
+	assertErrorJSON(t, stdout, "no_matching_work")
+}
+
+// TestListPlainTextOutput verifies list renders plain text rows for matching threads.
+func TestListPlainTextOutput(t *testing.T) {
+	t.Parallel()
+
+	dbPath := filepath.Join(t.TempDir(), "coord.db")
+	runInboxCommand(t, "--db", dbPath, "--json", "init")
+	threadID := sendPendingThread(t, dbPath, "leader", "worker-a", "Pending work", "Check latest failures")
+
+	stdout := runInboxCommand(
+		t,
+		"--db", dbPath,
+		"list",
+		"--assigned-to", "worker-a",
+		"--status", "pending",
+	)
+	lines := splitNonEmptyLines(stdout)
+	if len(lines) != 1 {
+		t.Fatalf("expected one plain text list row, got %#v", lines)
+	}
+	if !strings.Contains(lines[0], threadID+"\tpending\tworker-a\tPending work") {
+		t.Fatalf("expected thread %q in list output, got %q", threadID, lines[0])
+	}
+}
+
+// TestShowRejectsMarkReadWithoutAgent verifies show requires an agent for --mark-read.
+func TestShowRejectsMarkReadWithoutAgent(t *testing.T) {
+	t.Parallel()
+
+	dbPath := filepath.Join(t.TempDir(), "coord.db")
+	threadID := seedThreadForInboxTests(t, dbPath, "leader", "worker-a")
+
+	stdout, _, exitCode := executeInboxCommand(
+		"--db", dbPath,
+		"--json",
+		"show",
+		"--thread", threadID,
+		"--mark-read",
+	)
+	if exitCode != 30 {
+		t.Fatalf("expected invalid_input exit code 30, got %d\nstdout:\n%s", exitCode, stdout)
+	}
+	assertErrorJSON(t, stdout, "invalid_input")
+	assertQueryWaitErrorMessageContains(t, stdout, "agent is required when using --mark-read")
+}
+
+// TestShowPlainTextOutput verifies show renders plain text thread history including artifacts.
+func TestShowPlainTextOutput(t *testing.T) {
+	t.Parallel()
+
+	tempDir := t.TempDir()
+	dbPath := filepath.Join(tempDir, "coord.db")
+	artifactPath := filepath.Join(tempDir, "task.md")
+	if err := os.WriteFile(artifactPath, []byte("task brief"), 0o644); err != nil {
+		t.Fatalf("write artifact file: %v", err)
+	}
+	runInboxCommand(t, "--db", dbPath, "--json", "init")
+
+	sendOut := runInboxCommand(
+		t,
+		"--db", dbPath,
+		"--json",
+		"send",
+		"--from", "leader",
+		"--to", "worker-a",
+		"--subject", "Artifact task",
+		"--summary", "Attach brief",
+		"--artifact", artifactPath,
+		"--artifact-kind", "brief",
+	)
+	var sendResp map[string]any
+	mustDecodeJSON(t, sendOut, &sendResp)
+	threadID := nestedString(t, sendResp, "data", "thread", "thread_id")
+
+	stdout := runInboxCommand(t, "--db", dbPath, "show", "--thread", threadID)
+	if !strings.Contains(stdout, threadID+"\tpending\tArtifact task") {
+		t.Fatalf("expected thread header in show output, got %q", stdout)
+	}
+	if !strings.Contains(stdout, "- ") || !strings.Contains(stdout, "\ttask\tAttach brief") {
+		t.Fatalf("expected message row in show output, got %q", stdout)
+	}
+	if !strings.Contains(stdout, "  artifact\tbrief\t"+artifactPath) {
+		t.Fatalf("expected artifact row in show output, got %q", stdout)
+	}
+}
+
+// TestWatchUsesAfterEventResumeAndPlainTextOutput verifies watch resumes from after-event and renders plain text output.
+func TestWatchUsesAfterEventResumeAndPlainTextOutput(t *testing.T) {
+	t.Parallel()
+
+	dbPath := filepath.Join(t.TempDir(), "coord.db")
+	runInboxCommand(t, "--db", dbPath, "--json", "init")
+	threadID := sendPendingThread(t, dbPath, "leader", "worker-c", "Pending work", "Check latest failures")
+
+	firstWatchOut := runInboxCommand(
+		t,
+		"--db", dbPath,
+		"--json",
+		"watch",
+		"--agent", "worker-c",
+		"--status", "pending",
+		"--after-event", "0",
+		"--timeout-seconds", "1",
+	)
+	var firstWatchResp map[string]any
+	mustDecodeJSON(t, firstWatchOut, &firstWatchResp)
+	firstEventID := int64(nestedValue(t, firstWatchResp, "data", "next_event_id").(float64))
+
+	watchCh := make(chan watchCommandResult, 1)
+	go func() {
+		stdout, stderr, exitCode := executeInboxCommand(
+			"--db", dbPath,
+			"--agent", "worker-c",
+			"watch",
+			"--status", "blocked",
+			"--after-event", strconv.FormatInt(firstEventID, 10),
+			"--timeout-seconds", "2",
+		)
+		watchCh <- watchCommandResult{stdout: stdout, stderr: stderr, exit: exitCode}
+	}()
+
+	time.Sleep(200 * time.Millisecond)
+
+	runInboxCommand(t, "--db", dbPath, "--json", "claim", "--agent", "worker-c", "--thread", threadID)
+	runInboxCommand(
+		t,
+		"--db", dbPath,
+		"--json",
+		"update",
+		"--agent", "worker-c",
+		"--thread", threadID,
+		"--status", "blocked",
+		"--summary", "Need product decision",
+	)
+
+	select {
+	case result := <-watchCh:
+		if result.exit != 0 {
+			t.Fatalf("watch failed with exit=%d\nstderr:\n%s\nstdout:\n%s", result.exit, result.stderr, result.stdout)
+		}
+		if got := strings.TrimSpace(result.stdout); got != "watch woke on thread "+threadID+" at event "+strconv.FormatInt(firstEventID+2, 10) && !strings.Contains(got, "watch woke on thread "+threadID+" at event ") {
+			t.Fatalf("expected plain text wake output for %q, got %q", threadID, got)
+		}
+	case <-time.After(3 * time.Second):
+		t.Fatal("watch command did not return")
+	}
+}
+
+// TestWaitReplyUsesRootAgentAndPlainTextOutput verifies wait-reply works with the root agent and plain text output.
+func TestWaitReplyUsesRootAgentAndPlainTextOutput(t *testing.T) {
+	t.Parallel()
+
+	dbPath := filepath.Join(t.TempDir(), "coord.db")
+	threadID, blockedMessageID := seedBlockedThreadForWaitReply(t, dbPath)
+
+	waitCh := make(chan waitReplyCommandResult, 1)
+	go func() {
+		stdout, stderr, exitCode := executeInboxCommand(
+			"--db", dbPath,
+			"--agent", "worker-c",
+			"wait-reply",
+			"--thread", threadID,
+			"--after-message", blockedMessageID,
+			"--timeout-seconds", "2",
+		)
+		waitCh <- waitReplyCommandResult{stdout: stdout, stderr: stderr, exit: exitCode}
+	}()
+
+	time.Sleep(200 * time.Millisecond)
+
+	runInboxCommand(
+		t,
+		"--db", dbPath,
+		"--json",
+		"reply",
+		"--from", "leader",
+		"--to", "worker-c",
+		"--thread", threadID,
+		"--summary", "Use stdout",
+	)
+
+	select {
+	case result := <-waitCh:
+		if result.exit != 0 {
+			t.Fatalf("wait-reply failed with exit=%d\nstderr:\n%s\nstdout:\n%s", result.exit, result.stderr, result.stdout)
+		}
+		if got := strings.TrimSpace(result.stdout); !strings.Contains(got, "reply received on thread "+threadID+" at event ") {
+			t.Fatalf("expected plain text wait-reply output for %q, got %q", threadID, got)
+		}
+	case <-time.After(3 * time.Second):
+		t.Fatal("wait-reply command did not return")
+	}
+}
+
+// TestWaitReplyRejectsUnknownMessageCursor verifies wait-reply reports not_found for an unknown message cursor.
+func TestWaitReplyRejectsUnknownMessageCursor(t *testing.T) {
+	t.Parallel()
+
+	dbPath := initCommandTestDB(t)
+
+	stdout, _, exitCode := executeInboxCommand(
+		"--db", dbPath,
+		"--json",
+		"wait-reply",
+		"--thread", "thr_missing",
+		"--after-message", "msg_missing",
+	)
+	if exitCode != 40 {
+		t.Fatalf("expected not_found exit code 40, got %d\nstdout:\n%s", exitCode, stdout)
+	}
+	assertErrorJSON(t, stdout, "not_found")
+	assertQueryWaitErrorMessageContains(t, stdout, "message msg_missing not found in thread thr_missing")
+}
+
+func assertQueryWaitErrorMessageContains(t *testing.T, raw string, want string) {
+	t.Helper()
+
+	var payload map[string]any
+	mustDecodeJSON(t, raw, &payload)
+	errorValue, ok := payload["error"].(map[string]any)
+	if !ok {
+		t.Fatalf("expected error object, got %#v", payload["error"])
+	}
+	message, _ := errorValue["message"].(string)
+	if !strings.Contains(message, want) {
+		t.Fatalf("expected error message to contain %q, got %q", want, message)
+	}
+}
+
+func parsePlainThreadID(t *testing.T, stdout string, prefix string) string {
+	t.Helper()
+
+	line := strings.TrimSpace(stdout)
+	if !strings.HasPrefix(line, prefix) {
+		t.Fatalf("expected prefix %q in output %q", prefix, stdout)
+	}
+	threadID := strings.TrimSpace(strings.TrimPrefix(line, prefix))
+	if threadID == "" {
+		t.Fatalf("expected thread ID in output %q", stdout)
+	}
+	return threadID
+}
+
+func splitNonEmptyLines(stdout string) []string {
+	lines := strings.Split(strings.TrimSpace(stdout), "\n")
+	out := make([]string, 0, len(lines))
+	for _, line := range lines {
+		line = strings.TrimSpace(line)
+		if line != "" {
+			out = append(out, line)
+		}
+	}
+	return out
+}