From b34d03182255e4f3d87a4d13122f29ae0eb2e07a Mon Sep 17 00:00:00 2001 From: Fernando Gomes Date: Tue, 28 Jul 2026 15:52:54 -0300 Subject: [PATCH] feat(task-stream): publish pending wizard_ask prompts as pending_input While the agent is blocked on a wizard_ask question, a user who tabbed over to the PostHog web app has no way to know the run is waiting on them. Publish the open question on the task-stream payload so the web app's sync widget can surface it. The wire is a full-state upsert, so the signal is a nullable field, not an event: pending_input is set while a question is open and absent on the next push once it is answered, cancelled, or timed out. Question open/close transitions bypass the 250ms debounce, same as phase changes, since the point of the signal is immediacy. Sensitive asks (secrets) publish only the fact that input is required, never the prompt text. Backend side: PostHog/posthog#74324 Co-Authored-By: Claude Fable 5 --- .../__tests__/task-stream-push.test.ts | 99 ++++++++++++++++++- src/lib/task-stream/task-stream-push.ts | 39 +++++++- src/lib/task-stream/types.ts | 22 +++++ src/lib/wizard-ask-bridge.ts | 1 + src/lib/wizard-session.ts | 6 ++ 5 files changed, 163 insertions(+), 4 deletions(-) diff --git a/src/lib/task-stream/__tests__/task-stream-push.test.ts b/src/lib/task-stream/__tests__/task-stream-push.test.ts index 2e6155078..6accc51ef 100644 --- a/src/lib/task-stream/__tests__/task-stream-push.test.ts +++ b/src/lib/task-stream/__tests__/task-stream-push.test.ts @@ -5,7 +5,7 @@ import type { TaskStreamUpdate, } from '@lib/task-stream/types'; import type { WizardStore, TaskItem } from '@ui/tui/store'; -import { RunPhase } from '@lib/wizard-session'; +import { RunPhase, type PendingQuestion } from '@lib/wizard-session'; import { mkdtempSync, rmSync, writeFileSync } from 'node:fs'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; @@ -19,6 +19,7 @@ interface MockStoreState { tasks: TaskItem[]; eventPlan: unknown[]; installDir?: string; + pendingQuestion?: PendingQuestion | null; } function createMockStore(overrides: Partial = {}) { @@ -38,6 +39,7 @@ function createMockStore(overrides: Partial = {}) { skillId: state.skillId, outroData: null, installDir: state.installDir, + pendingQuestion: state.pendingQuestion ?? null, }; }, get tasks() { @@ -276,6 +278,58 @@ describe('TaskStreamPush', () => { expect(payload.skill_id).toMatch(/^[A-Za-z0-9_.-]+$/); }); + it('publishes pending_input while a wizard_ask is open', async () => { + const store = createMockStore({ + runPhase: RunPhase.Running, + pendingQuestion: pendingQuestion(), + }); + const { push, dest } = createPush(store); + + await push.push(); + + expect(dest.calls[0][1].pending_input).toEqual({ + id: 'q-1', + asked_at: '2026-07-28T10:00:00.000Z', + question_count: 1, + sensitive: false, + prompts: ['Which region is your project in?'], + }); + }); + + it('omits pending_input when no wizard_ask is open', async () => { + const store = createMockStore({ runPhase: RunPhase.Running }); + const { push, dest } = createPush(store); + + await push.push(); + + expect(dest.calls[0][1].pending_input).toBeUndefined(); + }); + + it('withholds prompts when any question is sensitive', async () => { + const store = createMockStore({ + runPhase: RunPhase.Running, + pendingQuestion: pendingQuestion({ + questions: [ + { + id: 'key', + prompt: 'Paste your API key', + kind: 'text', + sensitive: true, + }, + { id: 'region', prompt: 'Which region?', kind: 'text' }, + ], + }), + }); + const { push, dest } = createPush(store); + + await push.push(); + + const pendingInput = dest.calls[0][1].pending_input; + expect(pendingInput?.sensitive).toBe(true); + expect(pendingInput?.question_count).toBe(2); + expect(pendingInput?.prompts).toBeUndefined(); + }); + it('populates error when phase is Error', async () => { const store = createMockStore({ runPhase: RunPhase.Error }); const { push, dest } = createPush(store); @@ -444,6 +498,31 @@ describe('TaskStreamPush', () => { }); }); + describe('spec: wizard_ask open/close bypasses debounce', () => { + it('question appearing and resolving each produce an immediate push', async () => { + vi.useFakeTimers(); + const store = createMockStore({ runPhase: RunPhase.Running }); + const { push, dest } = createPush(store); + push.attach(); + + store._setAndEmit({ runPhase: RunPhase.Running }); + await flushMicrotasks(); + expect(dest.calls).toHaveLength(1); + + // wizard_ask opens — no debounce wait. + store._setAndEmit({ pendingQuestion: pendingQuestion() }); + await flushMicrotasks(); + expect(dest.calls).toHaveLength(2); + expect(dest.calls[1][1].pending_input?.id).toBe('q-1'); + + // User answers — the clearing push is just as immediate. + store._setAndEmit({ pendingQuestion: null }); + await flushMicrotasks(); + expect(dest.calls).toHaveLength(3); + expect(dest.calls[2][1].pending_input).toBeUndefined(); + }); + }); + describe('spec: coalesces concurrent emits during in-flight push', () => { it('emits during a slow flush produce one follow-up push with the latest state', async () => { const store = createMockStore({ runPhase: RunPhase.Running }); @@ -595,6 +674,24 @@ function taskItem(label: string): TaskItem { }; } +function pendingQuestion( + overrides: Partial = {}, +): PendingQuestion { + return { + id: 'q-1', + source: 'test-skill', + askedAt: '2026-07-28T10:00:00.000Z', + questions: [ + { + id: 'region', + prompt: 'Which region is your project in?', + kind: 'text', + }, + ], + ...overrides, + }; +} + function tasksUpTo(n: number): TaskItem[] { return Array.from({ length: n }, (_, i) => taskItem(`task-${i + 1}`)); } diff --git a/src/lib/task-stream/task-stream-push.ts b/src/lib/task-stream/task-stream-push.ts index 054004cd7..4994e6376 100644 --- a/src/lib/task-stream/task-stream-push.ts +++ b/src/lib/task-stream/task-stream-push.ts @@ -18,12 +18,18 @@ import type { WizardStore, TaskItem } from '@ui/tui/store'; import { TaskStatus } from '@ui/wizard-ui'; -import { RunPhase, OutroKind, type OutroData } from '@lib/wizard-session'; +import { + RunPhase, + OutroKind, + type OutroData, + type PendingQuestion, +} from '@lib/wizard-session'; import { type TaskStreamDestination, type TaskStreamUpdate, type StreamTask, type TaskStreamError, + type StreamPendingInput, StreamTaskStatus, StreamEvent, } from './types'; @@ -78,6 +84,25 @@ function buildError( return { type: 'wizard_error', message: 'Wizard run failed' }; } +/** + * Sensitive asks (secrets, API keys) publish only the fact that input is + * required — never the prompt text. The session row is team-visible and + * outlives the prompt. + */ +function buildPendingInput( + question: PendingQuestion | null | undefined, +): StreamPendingInput | undefined { + if (!question) return undefined; + const sensitive = question.questions.some((q) => q.sensitive === true); + return { + id: question.id, + asked_at: question.askedAt ?? new Date().toISOString(), + question_count: question.questions.length, + sensitive, + prompts: sensitive ? undefined : question.questions.map((q) => q.prompt), + }; +} + export interface TaskStreamPushOptions { store: WizardStore; programId: string; @@ -99,6 +124,7 @@ export class TaskStreamPush { private enabled: boolean; private created = false; private lastPushedPhase: RunPhase | null = null; + private lastPushedQuestionId: string | null = null; private unsubscribe: (() => void) | null = null; private debounceTimer: ReturnType | null = null; @@ -207,9 +233,14 @@ export class TaskStreamPush { } const phaseChanged = phase !== this.lastPushedPhase; - if (phaseChanged) { + const questionId = this.store.session.pendingQuestion?.id ?? null; + const questionChanged = questionId !== this.lastPushedQuestionId; + if (phaseChanged || questionChanged) { // Phase transitions bypass the debounce: the web app needs to - // see Running → Completed as soon as it lands. + // see Running → Completed as soon as it lands. A wizard_ask + // opening or closing is the same shape — the whole point of + // publishing it is that the user is looking at the web app, so + // a 250ms-stale "needs your input" defeats the purpose. if (this.debounceTimer) { clearTimeout(this.debounceTimer); this.debounceTimer = null; @@ -269,6 +300,7 @@ export class TaskStreamPush { tasks: buildTasks(tasks), event_plan: eventPlan.length > 0 ? { events: eventPlan } : undefined, error: buildError(phase, session.outroData), + pending_input: buildPendingInput(session.pendingQuestion), timestamp: new Date().toISOString(), }; @@ -285,6 +317,7 @@ export class TaskStreamPush { } this.lastPushedPhase = phase; + this.lastPushedQuestionId = payload.pending_input?.id ?? null; await Promise.all( this.destinations.map((d) => diff --git a/src/lib/task-stream/types.ts b/src/lib/task-stream/types.ts index bc5c504af..6f1c146aa 100644 --- a/src/lib/task-stream/types.ts +++ b/src/lib/task-stream/types.ts @@ -52,6 +52,27 @@ export interface StreamEventPlan { events: Array<{ name: string; description?: string }>; } +/** + * An in-flight `wizard_ask` request, published so the web app can tell + * the user their terminal is waiting on them. Null/absent means no input + * is pending — consumers treat the next push without this field as the + * dismissal signal. + */ +export interface StreamPendingInput { + /** PendingQuestion id — stable for the lifetime of one ask. */ + id: string; + /** UTC ISO 8601 timestamp of when the overlay opened. */ + asked_at: string; + question_count: number; + /** True when any question is marked sensitive (secrets, API keys). */ + sensitive: boolean; + /** + * Prompt texts, omitted entirely when `sensitive` is true so secret + * prompts never reach the team-visible session row. + */ + prompts?: string[]; +} + export interface TaskStreamUpdate { session_id: string; workflow_id: string; @@ -61,6 +82,7 @@ export interface TaskStreamUpdate { tasks: StreamTask[]; event_plan?: StreamEventPlan; error?: TaskStreamError; + pending_input?: StreamPendingInput; /** UTC ISO 8601 timestamp of this payload. Latest update wins on conflict. */ timestamp: string; } diff --git a/src/lib/wizard-ask-bridge.ts b/src/lib/wizard-ask-bridge.ts index 649aef828..a7dacaf3f 100644 --- a/src/lib/wizard-ask-bridge.ts +++ b/src/lib/wizard-ask-bridge.ts @@ -91,6 +91,7 @@ export function createWizardAskBridge( questions, source: opts.getSource(), richLinks: opts.richLinks ?? false, + askedAt: new Date().toISOString(), }; const startedAt = Date.now(); diff --git a/src/lib/wizard-session.ts b/src/lib/wizard-session.ts index 6b1a180fd..799099fbc 100644 --- a/src/lib/wizard-session.ts +++ b/src/lib/wizard-session.ts @@ -148,6 +148,12 @@ export type AskAnswers = Record; export interface PendingQuestion { id: string; questions: AskQuestion[]; + /** + * UTC ISO 8601 timestamp of when the ask was created. Published on the + * task stream as `pending_input.asked_at` so the web app can age the + * prompt; stable across pushes for the lifetime of one ask. + */ + askedAt?: string; /** Skill id of the caller. Set by the wizard from session.skillId. */ source: string; /**