| 1 | package control |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "path/filepath" |
| 7 | "testing" |
| 8 | |
| 9 | "reasonix/internal/event" |
| 10 | "reasonix/internal/mcpinteraction" |
| 11 | "reasonix/internal/turnevent" |
| 12 | ) |
| 13 | |
| 14 | func blockPromptTestLedger(t *testing.T, c *Controller, root string) { |
| 15 | t.Helper() |
| 16 | v3 := c.sessionEventStore() |
| 17 | if v3 == nil { |
| 18 | t.Fatal("controller did not open a v3 event store") |
| 19 | } |
| 20 | if err := v3.Close(context.Background()); err != nil { |
| 21 | t.Fatal(err) |
| 22 | } |
| 23 | } |
| 24 | |
| 25 | func awaitPromptLedgerTest[T any](t *testing.T, ch <-chan T, description string) T { |
| 26 | t.Helper() |
| 27 | select { |
| 28 | case result := <-ch: |
| 29 | return result |
| 30 | case <-t.Context().Done(): |
| 31 | t.Fatalf("test cancelled waiting for %s: %v", description, t.Context().Err()) |
| 32 | var zero T |
| 33 | return zero |
| 34 | } |
| 35 | } |
| 36 | |
| 37 | func TestCancelLedgerFailureReturnsAndCancelsTurn(t *testing.T) { |
| 38 | root := filepath.Join(t.TempDir(), "session-dir") |
| 39 | started := make(chan context.Context, 1) |
| 40 | finished := make(chan error, 1) |
| 41 | c := newOwnedTestController(t, Options{SessionDir: root, SessionPath: filepath.Join(root, "session.jsonl")}) |
| 42 | t.Cleanup(c.Close) |
| 43 | c.runGuarded(func(ctx context.Context) error { |
| 44 | started <- ctx |
| 45 | <-ctx.Done() |
| 46 | finished <- ctx.Err() |
| 47 | return ctx.Err() |
| 48 | }) |
| 49 | ctx := awaitPromptLedgerTest(t, started, "turn start") |
| 50 | blockPromptTestLedger(t, c, root) |
| 51 | returned := make(chan struct{}) |
| 52 | go func() { |
| 53 | c.Cancel() |
| 54 | close(returned) |
| 55 | }() |
| 56 | awaitPromptLedgerTest(t, returned, "Cancel to return after WAL failure") |
| 57 | awaitPromptLedgerTest(t, ctx.Done(), "turn context cancellation") |
| 58 | if err := awaitPromptLedgerTest(t, finished, "cancelled turn body"); !errors.Is(err, context.Canceled) { |
| 59 | t.Fatalf("turn error = %v, want context cancellation", err) |
| 60 | } |
| 61 | waitIdle(t, c) |
| 62 | if err := c.turnEventLedgerError(); !errors.Is(err, turnevent.ErrTurnLedgerUnavailable) { |
| 63 | t.Fatalf("ledger error = %v, want storage failure", err) |
| 64 | } |
| 65 | } |
| 66 | |
| 67 | func TestResolvePromptExactLedgerFailureCancelsWithoutAnswer(t *testing.T) { |
| 68 | tests := []struct { |
| 69 | name string |
| 70 | kind PromptKind |
| 71 | answer PromptAnswer |
| 72 | wait func(context.Context, *Controller) (bool, error) |
| 73 | }{ |
| 74 | { |
| 75 | name: "ask", kind: PromptAsk, |
| 76 | answer: PromptAnswer{Questions: []event.AskAnswer{{QuestionID: "q1", Selected: []string{"A"}}}}, |
| 77 | wait: func(ctx context.Context, c *Controller) (bool, error) { |
| 78 | answers, err := c.Ask(ctx, askProbeQuestions()) |
| 79 | return len(answers) != 0, err |
| 80 | }, |
| 81 | }, |
| 82 | { |
| 83 | name: "approval", kind: PromptApproval, answer: PromptAnswer{Allow: true}, |
| 84 | wait: func(ctx context.Context, c *Controller) (bool, error) { |
| 85 | allow, _, err := c.requestApprovalWithReason(ctx, "bash", "echo test", nil, "test") |
| 86 | return allow, err |
| 87 | }, |
| 88 | }, |
| 89 | { |
| 90 | name: "mcp", kind: PromptMCP, answer: PromptAnswer{Action: mcpinteraction.ActionAccept}, |
| 91 | wait: func(ctx context.Context, c *Controller) (bool, error) { |
| 92 | result, err := c.Interact(ctx, mcpinteraction.Request{Server: "test", Mode: "form", Message: "Confirm?"}) |
| 93 | return result.Action == mcpinteraction.ActionAccept, err |
| 94 | }, |
| 95 | }, |
| 96 | } |
| 97 | for _, tt := range tests { |
| 98 | t.Run(tt.name, func(t *testing.T) { |
| 99 | root := filepath.Join(t.TempDir(), "session-dir") |
| 100 | requests := make(chan event.Event, 1) |
| 101 | started := make(chan context.Context, 1) |
| 102 | type outcome struct { |
| 103 | answered bool |
| 104 | err error |
| 105 | } |
| 106 | finished := make(chan outcome, 1) |
| 107 | c := newOwnedTestController(t, Options{ |
| 108 | SessionDir: root, SessionPath: filepath.Join(root, "session.jsonl"), |
| 109 | Sink: event.FuncSink(func(e event.Event) { |
| 110 | switch e.Kind { |
| 111 | case event.AskRequest, event.ApprovalRequest, event.MCPInteractionRequest: |
| 112 | requests <- e |
| 113 | } |
| 114 | }), |
| 115 | }) |
| 116 | t.Cleanup(c.Close) |
| 117 | c.SetTurnEventRoutingMetadata("ledger-failure-test", "") |
| 118 | c.runGuarded(func(ctx context.Context) error { |
| 119 | started <- ctx |
| 120 | answered, err := tt.wait(ctx, c) |
| 121 | finished <- outcome{answered: answered, err: err} |
| 122 | return err |
| 123 | }) |
| 124 | ctx := awaitPromptLedgerTest(t, started, "turn start") |
| 125 | request := awaitPromptLedgerTest(t, requests, "prompt publication") |
| 126 | identity := PromptIdentity{PromptID: request.ItemID, TurnID: request.TurnID, RuntimeEpoch: "ledger-failure-test", Kind: tt.kind} |
| 127 | blockPromptTestLedger(t, c, root) |
| 128 | resolved := make(chan error, 1) |
| 129 | go func() { resolved <- c.ResolvePromptExact(identity, tt.answer) }() |
| 130 | if err := awaitPromptLedgerTest(t, resolved, "failed resolution to return"); !errors.Is(err, turnevent.ErrTurnLedgerUnavailable) { |
| 131 | t.Fatalf("ResolvePromptExact error = %v, want storage failure", err) |
| 132 | } |
| 133 | awaitPromptLedgerTest(t, ctx.Done(), "turn context cancellation") |
| 134 | result := awaitPromptLedgerTest(t, finished, "prompt waiter cancellation") |
| 135 | if result.answered || !errors.Is(result.err, context.Canceled) { |
| 136 | t.Fatalf("prompt outcome = %+v, want cancellation without an answer", result) |
| 137 | } |
| 138 | if pending := c.PendingPromptIdentities(); len(pending) != 0 { |
| 139 | t.Fatalf("failed turn left pending identities: %+v", pending) |
| 140 | } |
| 141 | if c.PendingPrompt() { |
| 142 | t.Fatal("failed turn left a pending approval or question") |
| 143 | } |
| 144 | if err := c.ResolvePromptExact(identity, tt.answer); err == nil { |
| 145 | t.Fatal("prompt accepted a later answer after ledger failure") |
| 146 | } |
| 147 | waitIdle(t, c) |
| 148 | }) |
| 149 | } |
| 150 | } |
| 151 |