ReviewOS
Reload

also looking at this

stacks/uplink

feat: let a text switch one thread between Claude and Codex

#52
Merged glennmichael123 wants to merge feat/per-thread-engine into main
16 files +316 -27

Review threads live on the whole diff, not on one commit, so none are shown here - a thread's line means something in the branch's final form, and painting it into an intermediate step would put it on code it is not about.

app/Commands/Uplink.tsmodified+1-0
@@ -79,6 +79,7 @@ export default defineCommand((cli) => {
7979 messages,
8080 sender,
8181 engine: selectedEngine(config),
82 engineFor: id => engineFor(id, config),
8283 store: new ModelStore(`${appDir}/storage/uplink/cursor`),
8384 })
8485 await uplink.start()
app/Uplink/commands.tsmodified+6-1
@@ -3,7 +3,7 @@
33 * message matches ("stop", "/stop", "Stop."), so "stop the dev server in
44 * ~/Code/x" still reaches the agent as a task.
55 */
6export type ControlCommand = 'help' | 'status' | 'stop' | 'new' | 'more' | 'ping'
6export type ControlCommand = 'help' | 'status' | 'stop' | 'new' | 'more' | 'ping' | 'claude' | 'codex'
77
88const ALIASES: Record<string, ControlCommand> = {
99 'help': 'help',
@@ -19,6 +19,10 @@ const ALIASES: Record<string, ControlCommand> = {
1919 'more': 'more',
2020 'continue reply': 'more',
2121 'ping': 'ping',
22 // Switching engines is a whole-message word like the rest, so "codex keeps
23 // timing out on ~/Code/x" is still a task and not a switch.
24 'claude': 'claude',
25 'codex': 'codex',
2226}
2327
2428export function parseControl(text: string): ControlCommand | null {
@@ -33,4 +37,5 @@ export const HELP_TEXT = [
3337 'new - start a fresh conversation',
3438 'more - rest of a long reply',
3539 'ping - check I am alive',
40 'claude / codex - switch this thread to that agent',
3641].join('\n')
app/Uplink/desktop-agent.tsmodified+4-2
@@ -8,7 +8,7 @@ import { homedir } from 'node:os'
88import { join } from 'node:path'
99import process from 'node:process'
1010import { loadConfig } from './config'
11import { allEngines, selectedEngine } from './engines'
11import { allEngines, engineFor, selectedEngine } from './engines'
1212import { formatDuration, truncate } from './format'
1313import { MessagesAccessError, MessagesDb } from './messages-db'
1414import { AppleScriptSender } from './sender'
@@ -300,7 +300,9 @@ export async function startDesktopAgent(options: { version: string }): Promise<D
300300 // license.
301301 if (!isLicensed(license))
302302 return
303 uplink = new Uplink({ config, messages, sender, engine: selectedEngine(config), store })
303 // `engineFor` as well as `engine`, so a thread that texts "codex" can
304 // reach the other one without restarting the agent.
305 uplink = new Uplink({ config, messages, sender, engine: selectedEngine(config), engineFor: id => engineFor(id, config), store })
304306 await uplink.start()
305307 }
306308
app/Uplink/uplink.tsmodified+63-6
@@ -1,6 +1,6 @@
11import type { ControlCommand } from './commands'
22import type { UplinkConfig } from './config'
3import type { Engine, EngineRun } from './engine'
3import type { Engine, EngineId, EngineRun } from './engine'
44import type { IncomingMessage } from './messages-db'
55import type { ReplyTarget, Sender } from './sender'
66import type { ConversationRecord, Store } from './store'
@@ -36,7 +36,13 @@ export interface UplinkDeps {
3636 config: UplinkConfig
3737 messages: MessagesSource
3838 sender: Sender
39 /** The installation's engine: what a thread uses until it says otherwise. */
3940 engine: Engine
41 /**
42 * How to reach the other engine when a thread texts its name. Left out, every
43 * thread gets `engine`, which is what a test with one fake wants.
44 */
45 engineFor?: (id: EngineId) => Engine
4046 store: Store
4147 systemPrompt?: string
4248 log?: Logger
@@ -48,6 +54,12 @@ interface Job {
4854 prompt: string
4955 cwd: string
5056 target: ReplyTarget
57 /**
58 * Fixed when the text arrived, not read when the run starts: a task queued
59 * behind another runs on the engine it was sent to, even if the thread
60 * switched while it waited.
61 */
62 engine: EngineId
5163}
5264
5365interface ActiveJob extends Job {
@@ -309,16 +321,18 @@ export class Uplink {
309321 const conversation = await this.conversation(target)
310322 const cwd = detectWorkdir(prompt, this.threadIsFresh(conversation) && conversation.cwd ? conversation.cwd : this.config.workdir)
311323
324 const engine = conversation.engine ?? this.config.engine
312325 const runId = await this.deps.store.createRun({
313326 chatGuid: target.chatGuid,
314327 messageGuid: message.guid,
315328 prompt,
316329 cwd,
317330 status: 'queued',
331 engine,
318332 })
319333
320334 const state = this.chat(target.chatGuid)
321 const job: Job = { runId, prompt, cwd, target }
335 const job: Job = { runId, prompt, cwd, target, engine }
322336 if (state.active) {
323337 state.queue.push(job)
324338 await this.reply(target, `Queued behind the current task (${state.queue.length} waiting). Text "status" or "stop".`)
@@ -328,19 +342,27 @@ export class Uplink {
328342 await this.begin(state, job)
329343 }
330344
345 /** The engine behind an id, falling back to the installation's one. */
346 private engineOf(id: EngineId): Engine {
347 return this.deps.engineFor?.(id) ?? this.deps.engine
348 }
349
331350 private threadIsFresh(conversation: ConversationRecord): boolean {
332351 return conversation.lastActiveAt !== null && this.now() - conversation.lastActiveAt < this.config.sessionIdleMs
333352 }
334353
335354 private async begin(state: ChatState, job: Job): Promise<void> {
336355 const conversation = await this.conversation(job.target)
337 const resume = conversation.sessionId && conversation.cwd === job.cwd && this.threadIsFresh(conversation)
356 // The engine has to agree as well: a session id belongs to the agent that
357 // made it, and handing a Claude one to Codex resumes nothing.
358 const sameEngine = (conversation.engine ?? this.config.engine) === job.engine
359 const resume = conversation.sessionId && conversation.cwd === job.cwd && sameEngine && this.threadIsFresh(conversation)
338360 ? conversation.sessionId
339361 : null
340362
341363 await this.deps.store.updateRun(job.runId, { status: 'running', startedAt: this.now() })
342364
343 const run = this.deps.engine.run({
365 const run = this.engineOf(job.engine).run({
344366 prompt: job.prompt,
345367 cwd: job.cwd,
346368 sessionId: resume,
@@ -407,7 +429,7 @@ export class Uplink {
407429 // Naming the wrong CLI, or a .env the downloadable app does not have,
408430 // sends someone at the Mac to fix something that is not broken. The
409431 // engine carries its own remedy.
410 const engine = this.deps.engine
432 const engine = this.engineOf(active.engine)
411433 const text = result.authFailure
412434 ? `I cannot reach ${engine.label}: it is not signed in on this Mac. Someone there needs to ${engine.authFailureHint}.`
413435 : result.ok ? result.text : `That failed: ${result.text}`
@@ -446,13 +468,19 @@ export class Uplink {
446468 return
447469
448470 case 'status': {
471 // A thread that can switch has to be able to say what it is on.
472 const conversation = await this.conversation(target)
473 const thread = this.engineOf(conversation.engine ?? this.config.engine).label
449474 if (!state.active) {
450 await this.reply(target, 'Idle. Text me a question or a task.')
475 await this.reply(target, `Idle on ${thread}. Text me a question or a task.`)
451476 return
452477 }
453478 const active = state.active
454479 const lines = [
455480 `Working ${formatDuration(this.now() - active.startedAt)} on: ${truncate(active.prompt, 80)}`,
481 // The running task's engine, which is not always the thread's: it
482 // keeps whatever it was sent to.
483 `Agent: ${this.engineOf(active.engine).label}`,
456484 active.lastActivity ? `Latest: ${active.lastActivity}` : 'Thinking.',
457485 ]
458486 if (state.queue.length > 0)
@@ -472,6 +500,35 @@ export class Uplink {
472500 return
473501 }
474502
503 case 'claude':
504 case 'codex': {
505 const conversation = await this.conversation(target)
506 const engine = this.engineOf(command)
507 if ((conversation.engine ?? this.config.engine) === command) {
508 await this.reply(target, `Already on ${engine.label}.`)
509 return
510 }
511
512 // Asked now rather than when the next text fails. Someone switching
513 // from a phone cannot see that the CLI is missing, and finding out on
514 // their real task costs them the task.
515 const probe = await engine.probe()
516 if (!probe.ok) {
517 await this.reply(target, probe.reason === 'missing'
518 ? `${engine.label} is not installed on this Mac. Install it there with: ${engine.install.command}`
519 : `${engine.label} is not ready: ${probe.detail}`)
520 return
521 }
522
523 // The session id goes with it. It belongs to the agent that made it,
524 // and the other one cannot resume it, so this starts a fresh
525 // conversation the way "new" does. The working directory stays: it is
526 // not the agent's, it is the thread's.
527 await this.deps.store.saveConversation({ ...conversation, engine: command, sessionId: null })
528 await this.reply(target, `Switched to ${engine.label}. Starting a fresh conversation, because a session cannot move between agents.`)
529 return
530 }
531
475532 case 'more': {
476533 const conversation = await this.conversation(target)
477534 if (!conversation.moreText) {
tests/unit/uplink/fixtures.tsmodified+20-7
@@ -1,4 +1,4 @@
1import type { Engine, EngineEvent, EngineProbe, EngineRequest, EngineResult, EngineRun } from '../../../app/Uplink/engine'
1import type { Engine, EngineEvent, EngineId, EngineInstall, EngineProbe, EngineRequest, EngineResult, EngineRun } from '../../../app/Uplink/engine'
22import type { AutomationState, ReplyTarget, Sender } from '../../../app/Uplink/sender'
33import { Database } from 'bun:sqlite'
44import { mkdtempSync } from 'node:fs'
@@ -153,12 +153,25 @@ export class FakeEngine implements Engine {
153153 runs: PendingRun[] = []
154154 probes = 0
155155
156 // Identity matching the real Claude engine, so a test that asserts on a
157 // reply naming the CLI is asserting the text a user would actually get.
158 readonly id = 'claude' as const
159 readonly label = 'Claude Code'
160 readonly authFailureHint = 'run "claude setup-token" on the Mac and give Uplink the token'
161 readonly install = { command: 'bun install -g @anthropic-ai/claude-code', url: 'https://docs.claude.com/en/docs/claude-code' }
156 // Identity matching the real engines, so a test that asserts on a reply
157 // naming the CLI is asserting the text a user would actually get. Defaults
158 // to Claude, which is what a test with one engine wants.
159 readonly id: EngineId
160 readonly label: string
161 readonly authFailureHint: string
162 readonly install: EngineInstall
163
164 constructor(id: EngineId = 'claude') {
165 this.id = id
166 const claude = id === 'claude'
167 this.label = claude ? 'Claude Code' : 'Codex'
168 this.authFailureHint = claude
169 ? 'run "claude setup-token" on the Mac and give Uplink the token'
170 : 'run "codex login" on the Mac (or "codex login --device-auth" over SSH)'
171 this.install = claude
172 ? { command: 'bun install -g @anthropic-ai/claude-code', url: 'https://docs.claude.com/en/docs/claude-code' }
173 : { command: 'bun install -g @openai/codex', url: 'https://developers.openai.com/codex/cli' }
174 }
162175
163176 probeResult: EngineProbe = { ok: true, reason: 'ok', detail: 'Signed in' }
164177
tests/unit/uplink/uplink.test.tsmodified+141-3
@@ -20,9 +20,13 @@ function config(overrides: Partial<UplinkConfig> = {}): UplinkConfig {
2020 progressEveryMs: 0,
2121 sessionIdleMs: 6 * 3_600_000,
2222 workdir: '/tmp',
23 engine: 'claude',
2324 claudeBin: 'claude',
24 model: null,
25 claudeModel: null,
2526 permissionMode: 'bypassPermissions',
27 codexBin: 'codex',
28 codexModel: null,
29 codexPermission: 'bypass',
2630 timeoutMs: 60_000,
2731 ...overrides,
2832 }
@@ -32,6 +36,8 @@ interface Harness {
3236 fake: FakeChatDb
3337 sender: FakeSender
3438 engine: FakeEngine
39 /** The other agent, reachable once a thread texts its name. */
40 codex: FakeEngine
3541 store: MemoryStore
3642 uplink: Uplink
3743}
@@ -44,10 +50,19 @@ async function harness(overrides: Partial<UplinkConfig> = {}, seed?: (fake: Fake
4450 seed?.(fake)
4551 const sender = new FakeSender(fake)
4652 const engine = new FakeEngine()
53 const codex = new FakeEngine('codex')
4754 const store = new MemoryStore()
48 const uplink = new Uplink({ config: config(overrides), messages: new MessagesDb(fake.path), sender, engine, store, log: silent })
55 const uplink = new Uplink({
56 config: config(overrides),
57 messages: new MessagesDb(fake.path),
58 sender,
59 engine,
60 engineFor: id => (id === 'codex' ? codex : engine),
61 store,
62 log: silent,
63 })
4964 await uplink.start({ poll: false }) // Each test drives tick() itself.
50 return { fake, sender, engine, store, uplink }
65 return { fake, sender, engine, codex, store, uplink }
5166}
5267
5368afterEach(() => open?.close())
@@ -243,4 +258,127 @@ describe('Uplink', () => {
243258 expect(h.sender.texts()[0]).toContain('claude setup-token')
244259 expect(h.store.runs.get(1)?.status).toBe('failed')
245260 })
261
262 /**
263 * #10: a global setting cannot answer "use Codex for this one thread" from a
264 * phone, which is the point of the product.
265 */
266 it('switches one thread to the other agent, and does not carry the session across', async () => {
267 const h = await harness()
268 h.fake.add({ chat: ME, text: 'hello', fromMe: true })
269 await h.uplink.tick()
270 h.engine.finish(0, 'hi', { sessionId: 'claude-session-1' })
271 await settle()
272
273 h.fake.add({ chat: ME, text: 'codex', fromMe: true })
274 await h.uplink.tick()
275 expect(h.sender.texts().at(-1)).toContain('Switched to Codex')
276
277 h.fake.add({ chat: ME, text: 'now do the thing', fromMe: true })
278 await h.uplink.tick()
279
280 // The task went to the other agent, and Claude was not asked again.
281 expect(h.codex.runs).toHaveLength(1)
282 expect(h.engine.runs).toHaveLength(1)
283 // A Claude session id cannot be resumed by Codex, so it is not offered.
284 expect(h.codex.runs[0].request.sessionId).toBeNull()
285 expect(h.store.runs.get(2)?.engine).toBe('codex')
286 })
287
288 it('keeps the thread on the new agent, and switches back when asked', async () => {
289 const h = await harness()
290 h.fake.add({ chat: ME, text: 'codex', fromMe: true })
291 await h.uplink.tick()
292 h.fake.add({ chat: ME, text: 'one', fromMe: true })
293 await h.uplink.tick()
294 h.codex.finish(0, 'done')
295 await settle()
296
297 // Still Codex without being told again.
298 h.fake.add({ chat: ME, text: 'two', fromMe: true })
299 await h.uplink.tick()
300 expect(h.codex.runs).toHaveLength(2)
301 h.codex.finish(1, 'done')
302 await settle()
303
304 h.fake.add({ chat: ME, text: 'claude', fromMe: true })
305 await h.uplink.tick()
306 expect(h.sender.texts().at(-1)).toContain('Switched to Claude Code')
307 h.fake.add({ chat: ME, text: 'three', fromMe: true })
308 await h.uplink.tick()
309 expect(h.engine.runs).toHaveLength(1)
310 })
311
312 it('says so rather than switching twice', async () => {
313 const h = await harness()
314 h.fake.add({ chat: ME, text: 'claude', fromMe: true })
315 await h.uplink.tick()
316 expect(h.sender.texts().at(-1)).toContain('Already on Claude Code')
317 })
318
319 /**
320 * A task queued behind another runs on the agent it was sent to. Reading the
321 * thread's engine at start time instead would silently move a waiting task
322 * onto an agent its author never chose.
323 */
324 it('runs a queued task on the agent it was sent to', async () => {
325 const h = await harness()
326 h.fake.add({ chat: ME, text: 'first', fromMe: true })
327 await h.uplink.tick()
328 h.fake.add({ chat: ME, text: 'second', fromMe: true })
329 await h.uplink.tick() // queued behind the first
330
331 h.fake.add({ chat: ME, text: 'codex', fromMe: true })
332 await h.uplink.tick()
333
334 h.engine.finish(0, 'first done')
335 await settle()
336
337 // The queued one still went to Claude, and its row says so.
338 expect(h.engine.runs).toHaveLength(2)
339 expect(h.codex.runs).toHaveLength(0)
340 expect(h.store.runs.get(2)?.engine).toBe('claude')
341 })
342
343 it('will not switch to an agent that is not installed', async () => {
344 const h = await harness()
345 h.codex.probeResult = { ok: false, reason: 'missing', detail: 'Not found' }
346 h.fake.add({ chat: ME, text: 'codex', fromMe: true })
347 await h.uplink.tick()
348
349 expect(h.sender.texts().at(-1)).toContain('not installed')
350 // Told how, because nobody switching from a phone can see the Mac.
351 expect(h.sender.texts().at(-1)).toContain('bun install -g @openai/codex')
352
353 // And the thread did not move.
354 h.fake.add({ chat: ME, text: 'a task', fromMe: true })
355 await h.uplink.tick()
356 expect(h.engine.runs).toHaveLength(1)
357 })
358
359 it('says which agent an idle thread is on', async () => {
360 const h = await harness()
361 h.fake.add({ chat: ME, text: 'status', fromMe: true })
362 await h.uplink.tick()
363 expect(h.sender.texts().at(-1)).toContain('Idle on Claude Code')
364 })
365
366 it('names the agent the running task is on', async () => {
367 const h = await harness()
368 h.fake.add({ chat: ME, text: 'codex', fromMe: true })
369 await h.uplink.tick()
370 h.fake.add({ chat: ME, text: 'a task', fromMe: true })
371 await h.uplink.tick()
372 h.fake.add({ chat: ME, text: 'status', fromMe: true })
373 await h.uplink.tick()
374 expect(h.sender.texts().at(-1)).toContain('Agent: Codex')
375 })
376
377 it('offers both words in help', async () => {
378 const h = await harness()
379 h.fake.add({ chat: ME, text: 'help', fromMe: true })
380 await h.uplink.tick()
381 expect(h.sender.texts().at(-1)).toContain('claude / codex')
382 })
246383})
384