| 1 | package control |
| 2 | |
| 3 | import ( |
| 4 | "errors" |
| 5 | "os" |
| 6 | "path/filepath" |
| 7 | "testing" |
| 8 | "time" |
| 9 | |
| 10 | "reasonix/internal/agent" |
| 11 | "reasonix/internal/event" |
| 12 | "reasonix/internal/sessioninbox" |
| 13 | "reasonix/internal/tool" |
| 14 | ) |
| 15 | |
| 16 | func TestSteerEventFollowsDurableConsumedTransition(t *testing.T) { |
| 17 | dir := t.TempDir() |
| 18 | prov := &inboxSteerProvider{started: make(chan struct{}), release: make(chan struct{})} |
| 19 | exec := agent.New(prov, tool.NewRegistry(), agent.NewSession("sys"), agent.Options{}, event.Discard) |
| 20 | observed := make(chan sessioninbox.InboxState, 1) |
| 21 | done := make(chan struct{}) |
| 22 | var c *Controller |
| 23 | sink := event.FuncSink(func(e event.Event) { |
| 24 | if e.Kind == event.Steer && c != nil { |
| 25 | state := sessioninbox.InboxState("") |
| 26 | for _, item := range c.InboxSnapshot().Items { |
| 27 | if item.ID == e.ItemID { |
| 28 | state = item.State |
| 29 | break |
| 30 | } |
| 31 | } |
| 32 | observed <- state |
| 33 | } |
| 34 | if e.Kind == event.TurnDone { |
| 35 | select { |
| 36 | case <-done: |
| 37 | default: |
| 38 | close(done) |
| 39 | } |
| 40 | } |
| 41 | }) |
| 42 | c = newOwnedTestController(t, Options{ |
| 43 | Runner: exec, |
| 44 | Executor: exec, |
| 45 | Sink: sink, |
| 46 | SessionDir: dir, |
| 47 | SessionPath: filepath.Join(dir, "s.jsonl"), |
| 48 | }) |
| 49 | t.Cleanup(func() { |
| 50 | c.Close() |
| 51 | c.autosaveWG.Wait() |
| 52 | }) |
| 53 | c.Submit("initial turn") |
| 54 | select { |
| 55 | case <-prov.started: |
| 56 | case <-time.After(time.Second): |
| 57 | t.Fatal("initial provider turn did not start") |
| 58 | } |
| 59 | rec, err := c.EnqueueInbox(InboxRequest{Intent: sessioninbox.IntentSteer, Submit: "durable steer"}) |
| 60 | if err != nil { |
| 61 | t.Fatal(err) |
| 62 | } |
| 63 | if _, err := c.TrySteerInboxItem(rec.ItemID); err != nil { |
| 64 | t.Fatal(err) |
| 65 | } |
| 66 | close(prov.release) |
| 67 | select { |
| 68 | case state := <-observed: |
| 69 | if state != sessioninbox.StateSteerConsumed { |
| 70 | t.Fatalf("state at steer event = %q, want %q", state, sessioninbox.StateSteerConsumed) |
| 71 | } |
| 72 | case <-time.After(5 * time.Second): |
| 73 | t.Fatal("timed out waiting for steer event") |
| 74 | } |
| 75 | select { |
| 76 | case <-done: |
| 77 | case <-time.After(5 * time.Second): |
| 78 | t.Fatal("timed out waiting for turn completion") |
| 79 | } |
| 80 | } |
| 81 | |
| 82 | func TestCancelWithInboxItemsResultRestoresOnlyUnconsumedItems(t *testing.T) { |
| 83 | dir := t.TempDir() |
| 84 | session := filepath.Join(dir, "s.jsonl") |
| 85 | _ = os.WriteFile(session, []byte("{}\n"), 0o644) |
| 86 | c := newOwnedTestController(t, Options{SessionPath: session, SessionDir: dir, Sink: event.Discard}) |
| 87 | accepted, err := c.EnqueueInbox(InboxRequest{Submit: "accepted", Source: "desktop"}) |
| 88 | if err != nil { |
| 89 | t.Fatal(err) |
| 90 | } |
| 91 | consumed, err := c.EnqueueInbox(InboxRequest{Submit: "consumed", Source: "desktop"}) |
| 92 | if err != nil { |
| 93 | t.Fatal(err) |
| 94 | } |
| 95 | st, err := c.ensureInbox() |
| 96 | if err != nil { |
| 97 | t.Fatal(err) |
| 98 | } |
| 99 | if err := st.SetState(accepted.ItemID, sessioninbox.StateSteerAccepted, ""); err != nil { |
| 100 | t.Fatal(err) |
| 101 | } |
| 102 | if err := st.SetState(consumed.ItemID, sessioninbox.StateSteerConsumed, ""); err != nil { |
| 103 | t.Fatal(err) |
| 104 | } |
| 105 | |
| 106 | result, err := c.CancelWithInboxItemsResult([]string{accepted.ItemID, consumed.ItemID}, "desktop") |
| 107 | if err != nil { |
| 108 | t.Fatal(err) |
| 109 | } |
| 110 | if len(result.DiscardedItemIDs) != 1 || result.DiscardedItemIDs[0] != accepted.ItemID { |
| 111 | t.Fatalf("discarded ids = %v", result.DiscardedItemIDs) |
| 112 | } |
| 113 | items := c.InboxSnapshot().Items |
| 114 | if len(items) != 1 || items[0].ID != consumed.ItemID { |
| 115 | t.Fatalf("remaining items = %+v", items) |
| 116 | } |
| 117 | } |
| 118 | |
| 119 | func TestDeleteInboxItemDoesNotOverwriteConsumedSteer(t *testing.T) { |
| 120 | dir := t.TempDir() |
| 121 | session := filepath.Join(dir, "s.jsonl") |
| 122 | _ = os.WriteFile(session, []byte("{}\n"), 0o644) |
| 123 | c := newOwnedTestController(t, Options{SessionPath: session, SessionDir: dir, Sink: event.Discard}) |
| 124 | rec, err := c.EnqueueInbox(InboxRequest{Intent: sessioninbox.IntentSteer, Submit: "consumed"}) |
| 125 | if err != nil { |
| 126 | t.Fatal(err) |
| 127 | } |
| 128 | st, err := c.ensureInbox() |
| 129 | if err != nil { |
| 130 | t.Fatal(err) |
| 131 | } |
| 132 | if err := st.SetState(rec.ItemID, sessioninbox.StateSteerAccepted, ""); err != nil { |
| 133 | t.Fatal(err) |
| 134 | } |
| 135 | c.inbox.mu.Lock() |
| 136 | c.inbox.trackActive(rec.ItemID) |
| 137 | c.inbox.mu.Unlock() |
| 138 | if err := st.MarkSteerConsumed(rec.ItemID); err != nil { |
| 139 | t.Fatal(err) |
| 140 | } |
| 141 | if err := c.DeleteInboxItem(rec.ItemID); !errors.Is(err, sessioninbox.ErrInvalidState) { |
| 142 | t.Fatalf("delete consumed steer = %v, want ErrInvalidState", err) |
| 143 | } |
| 144 | meta, _, err := c.ReadInboxItem(rec.ItemID) |
| 145 | if err != nil { |
| 146 | t.Fatal(err) |
| 147 | } |
| 148 | if meta.State != sessioninbox.StateSteerConsumed { |
| 149 | t.Fatalf("state = %q, want %q", meta.State, sessioninbox.StateSteerConsumed) |
| 150 | } |
| 151 | } |
| 152 | |
| 153 | type inboxChangedCapture struct { |
| 154 | changed chan sessioninbox.InboxSnapshot |
| 155 | } |
| 156 | |
| 157 | func (s *inboxChangedCapture) Emit(event.Event) {} |
| 158 | |
| 159 | func (s *inboxChangedCapture) InboxChanged(snap sessioninbox.InboxSnapshot) { |
| 160 | s.changed <- snap |
| 161 | } |
| 162 | |
| 163 | func TestInboxStoreChangesReachOptionalSink(t *testing.T) { |
| 164 | dir := t.TempDir() |
| 165 | sink := &inboxChangedCapture{changed: make(chan sessioninbox.InboxSnapshot, 1)} |
| 166 | c := newOwnedTestController(t, Options{ |
| 167 | SessionPath: filepath.Join(dir, "s.jsonl"), |
| 168 | SessionDir: dir, |
| 169 | Sink: sink, |
| 170 | }) |
| 171 | rec, err := c.EnqueueInbox(InboxRequest{Submit: "notify", Source: "desktop"}) |
| 172 | if err != nil { |
| 173 | t.Fatal(err) |
| 174 | } |
| 175 | awaitState := func(want sessioninbox.InboxState) sessioninbox.InboxSnapshot { |
| 176 | t.Helper() |
| 177 | select { |
| 178 | case snap := <-sink.changed: |
| 179 | if len(snap.Items) != 1 || snap.Items[0].State != want { |
| 180 | t.Fatalf("notification = %+v, want state %q", snap.Items, want) |
| 181 | } |
| 182 | return snap |
| 183 | case <-time.After(time.Second): |
| 184 | t.Fatalf("timed out waiting for %q notification", want) |
| 185 | return sessioninbox.InboxSnapshot{} |
| 186 | } |
| 187 | } |
| 188 | queued := awaitState(sessioninbox.StateQueued) |
| 189 | st, err := c.ensureInbox() |
| 190 | if err != nil { |
| 191 | t.Fatal(err) |
| 192 | } |
| 193 | if err := st.SetState(rec.ItemID, sessioninbox.StateSteerAccepted, ""); err != nil { |
| 194 | t.Fatal(err) |
| 195 | } |
| 196 | accepted := awaitState(sessioninbox.StateSteerAccepted) |
| 197 | if err := st.MarkSteerConsumed(rec.ItemID); err != nil { |
| 198 | t.Fatal(err) |
| 199 | } |
| 200 | consumed := awaitState(sessioninbox.StateSteerConsumed) |
| 201 | if !(queued.Revision < accepted.Revision && accepted.Revision < consumed.Revision) { |
| 202 | t.Fatalf("revisions did not increase: %d, %d, %d", queued.Revision, accepted.Revision, consumed.Revision) |
| 203 | } |
| 204 | } |
| 205 |