diff --git a/src/databases/officer_db/src/index.ts b/src/databases/officer_db/src/index.ts index 957b50b6..06e93ac9 100644 --- a/src/databases/officer_db/src/index.ts +++ b/src/databases/officer_db/src/index.ts @@ -35,6 +35,7 @@ export * from './permissions'; export * from './chat-events'; export * from './chat-session-models'; export * from './chat-notification-prefs'; +export * from './nudges'; export * from './ai-provider-usage'; export * from './command-hq'; export * from './dashboards'; diff --git a/src/databases/officer_db/src/nudges/index.ts b/src/databases/officer_db/src/nudges/index.ts new file mode 100644 index 00000000..1527547b --- /dev/null +++ b/src/databases/officer_db/src/nudges/index.ts @@ -0,0 +1 @@ +export { setNudge, getNudge, cancelNudge, dueNudges, recordNudgeSent } from './queries'; diff --git a/src/databases/officer_db/src/nudges/queries.ts b/src/databases/officer_db/src/nudges/queries.ts new file mode 100644 index 00000000..e2e09aff --- /dev/null +++ b/src/databases/officer_db/src/nudges/queries.ts @@ -0,0 +1,80 @@ +import { and, eq, lt, or, isNull, sql } from 'drizzle-orm'; +import { db } from '../db'; +import { nudges } from './schema'; +import type { NudgeSelect } from '../types'; + +/** Create or replace the schedule for one session. Re-arming resets the count — it is a new night. */ +export async function setNudge(params: { + userId: number; + sessionId: string; + message: string; + everySeconds: number; + maxCount: number; +}): Promise { + const [row] = await db + .insert(nudges) + .values({ ...params, enabled: true, sentCount: 0, lastFiredAt: null }) + .onConflictDoUpdate({ + target: [nudges.userId, nudges.sessionId], + set: { + message: params.message, + everySeconds: params.everySeconds, + maxCount: params.maxCount, + enabled: true, + sentCount: 0, + lastFiredAt: null, + updatedAt: new Date(), + }, + }) + .returning(); + return row!; +} + +export async function getNudge(userId: number, sessionId: string): Promise { + const [row] = await db + .select() + .from(nudges) + .where(and(eq(nudges.userId, userId), eq(nudges.sessionId, sessionId))) + .limit(1); + return row ?? null; +} + +export async function cancelNudge(userId: number, sessionId: string): Promise { + await db.delete(nudges).where(and(eq(nudges.userId, userId), eq(nudges.sessionId, sessionId))); +} + +/** + * Everything the scheduler should consider this tick. + * + * Due means: enabled, under its ceiling, and either never fired or last fired at least `every_seconds` + * ago. Whether the session is actually IDLE is not asked here — that lives in the sidecar, not the + * database, and the scheduler checks it immediately before speaking. + */ +export async function dueNudges(now: Date = new Date()): Promise { + return db + .select() + .from(nudges) + .where( + and( + eq(nudges.enabled, true), + lt(nudges.sentCount, nudges.maxCount), + or( + isNull(nudges.lastFiredAt), + lt(nudges.lastFiredAt, sql`${now} - (${nudges.everySeconds} * interval '1 second')`), + ), + ), + ); +} + +/** One nudge delivered. Disables itself on reaching the ceiling rather than leaving a spent row armed. */ +export async function recordNudgeSent(id: number): Promise { + await db + .update(nudges) + .set({ + sentCount: sql`${nudges.sentCount} + 1`, + lastFiredAt: new Date(), + updatedAt: new Date(), + enabled: sql`${nudges.sentCount} + 1 < ${nudges.maxCount}`, + }) + .where(eq(nudges.id, id)); +} diff --git a/src/databases/officer_db/src/nudges/schema.ts b/src/databases/officer_db/src/nudges/schema.ts new file mode 100644 index 00000000..40f05dd2 --- /dev/null +++ b/src/databases/officer_db/src/nudges/schema.ts @@ -0,0 +1,42 @@ +import { pgTable, serial, integer, text, boolean, timestamp, check, uniqueIndex } from 'drizzle-orm/pg-core'; +import { sql } from 'drizzle-orm'; +import { users } from '../auth/schema'; + +// Nudger: keep a chat moving while nobody is watching it. +// +// An agent told to record blockers and continue will still stop at the end of a turn, and at 3am nobody +// types the next message. This is the schedule that does. +// +// One nudge per (user, session). A second schedule for the same conversation is not a feature — it is two +// alarms for one sleeper, and the only thing it reliably produces is a doubled prompt. + +export const nudges = pgTable( + 'nudges', + { + id: serial('id').primaryKey(), + userId: integer('user_id') + .notNull() + .references(() => users.id, { onDelete: 'cascade' }), + /** The chat session's transcript uuid — the same key `chat_session_models` uses. */ + sessionId: text('session_id').notNull(), + message: text('message').notNull(), + everySeconds: integer('every_seconds').notNull().default(3600), + /** A loop nobody is watching should end on its own. */ + maxCount: integer('max_count').notNull().default(12), + sentCount: integer('sent_count').notNull().default(0), + enabled: boolean('enabled').notNull().default(true), + /** Null until the first send. The scheduler treats null as "due now". */ + lastFiredAt: timestamp('last_fired_at', { withTimezone: true }), + createdAt: timestamp('created_at', { withTimezone: true }).notNull().defaultNow(), + updatedAt: timestamp('updated_at', { withTimezone: true }).notNull().defaultNow(), + }, + (table) => [ + // uniqueIndex, not unique().on() — drizzle-kit re-creates named composite CONSTRAINTS on every push. + // See src/databases/CLAUDE.md → "Composite keys". + uniqueIndex('uq_nudges_user_session').on(table.userId, table.sessionId), + // A floor, because the scheduler ticks once a minute and anything under that is a lie about cadence + // rather than a faster nudge. + check('ck_nudges_every_seconds', sql`${table.everySeconds} >= 60`), + check('ck_nudges_max_count', sql`${table.maxCount} > 0`), + ], +); diff --git a/src/databases/officer_db/src/schema.ts b/src/databases/officer_db/src/schema.ts index 0a36ce10..de9e9f76 100644 --- a/src/databases/officer_db/src/schema.ts +++ b/src/databases/officer_db/src/schema.ts @@ -45,6 +45,7 @@ export * from './app-store/schema'; // sidecar_installs export * from './plugins/schema'; // plugin_installs — core: the plugin system is the platform's own export * from './service-connections/schema'; // service_connections export * from './notify/schema'; // push_devices — core since 2026-08-19, see docs/push-notifications.md +export * from './nudges/schema'; // nudges — the Nudger schedule // ── Plugins — uncomment when the plugin is installed ───────────────────────────────────────────── diff --git a/src/databases/officer_db/src/types.ts b/src/databases/officer_db/src/types.ts index 4bf0f2ca..9494989b 100644 --- a/src/databases/officer_db/src/types.ts +++ b/src/databases/officer_db/src/types.ts @@ -82,6 +82,7 @@ export type PipelineJobSelect = typeof Schema.pipelineJobs.$inferSelect; export type PipelineJobInsert = typeof Schema.pipelineJobs.$inferInsert; // Notifications +export type NudgeSelect = typeof Schema.nudges.$inferSelect; export type PushDeviceSelect = typeof NotifySchema.pushDevices.$inferSelect; export type PushDeviceInsert = typeof NotifySchema.pushDevices.$inferInsert; diff --git a/src/server.tsx b/src/server.tsx index 5aab2447..f0d69c8e 100644 --- a/src/server.tsx +++ b/src/server.tsx @@ -19,6 +19,7 @@ import { terminalWebsocket } from './servers/api/terminal/websocket'; import { chatWebsocket } from './servers/api/chat/websocket'; import { chatNotifyWebsocket } from './servers/api/chat/notify-delivery'; import { registerChatPushNotifications } from './servers/api/chat/notify-push'; +import { startNudger } from './servers/api/chat/nudger'; import { taskRunnerWebsocket } from './servers/api/tasks/task-executor'; import { pipelineWebsocket } from './servers/api/tasks/pipeline-executor'; // import { desktopWebsocket } from './servers/api/desktop/websocket'; @@ -166,6 +167,10 @@ assertPermissionTotality({ // module whose only job is a side effect is one an import sweep quietly deletes. registerChatPushNotifications(); +// Nudger's tick. Here rather than on first use, because a schedule armed before a restart has to survive +// one — the row is in Postgres and nothing else would ever look at it again. +startNudger(); + // That the working directory is the repo, because every path derives from its parent. First of the three, // since a wrong answer here makes the other two check the wrong files. assertInstallLayout(); diff --git a/src/servers/api/chat/chat.ts b/src/servers/api/chat/chat.ts index 680409fb..66c4e018 100644 --- a/src/servers/api/chat/chat.ts +++ b/src/servers/api/chat/chat.ts @@ -14,6 +14,7 @@ import { setChatNotifyPreference, } from 'officerdb'; import type { NotifyPreference } from 'officerdb/types'; +import { setNudge, getNudge, cancelNudge } from 'officerdb'; import { listClaudePwds, listClaudeSessions, @@ -548,3 +549,52 @@ chatRouter.post('/stt', async (ctx: Context) => { // Lives on this router because it is the same authority `chat` already grants: creating and naming // Claude sessions. See servers/api/chat/agent-panels-routes.ts and docs/agent-coordination.md. registerAgentPanelRoutes(chatRouter); + +// ── Nudger ── +// +// A schedule that keeps one chat moving while nobody is watching it. The tick lives in ./nudger.ts; these +// are only the three verbs the composer needs. +// +// Scoped to the caller like every other per-session route here. A nudge is a prompt sent AS someone, so a +// schedule readable or writable across accounts would be a way to speak in another person's chat. + +const MIN_EVERY_SECONDS = 60; +const MAX_EVERY_SECONDS = 86_400; + +// GET /chat/sessions/:id/nudge — the current schedule, or null. +chatRouter.get('/sessions/:id/nudge', async (ctx) => { + const user = ctx.get('user'); + await chatIdentity(user); + return ctx.json({ nudge: await getNudge(user.id, ctx.req.param('id')) }); +}); + +// PUT /chat/sessions/:id/nudge — arm or re-arm. Re-arming resets the count: it is a new night. +chatRouter.put('/sessions/:id/nudge', async (ctx) => { + const user = ctx.get('user'); + await chatIdentity(user); + + const body = await ctx.req.json<{ message?: string; everySeconds?: number; maxCount?: number }>(); + const message = typeof body.message === 'string' ? body.message.trim() : ''; + if (!message) return ctx.text('message is required', 400); + + const everySeconds = Number(body.everySeconds ?? 3600); + const maxCount = Number(body.maxCount ?? 12); + // The floor is the tick interval — anything under it is a lie about cadence rather than a faster nudge. + if (!Number.isFinite(everySeconds) || everySeconds < MIN_EVERY_SECONDS || everySeconds > MAX_EVERY_SECONDS) { + return ctx.text(`everySeconds must be between ${MIN_EVERY_SECONDS} and ${MAX_EVERY_SECONDS}`, 400); + } + if (!Number.isFinite(maxCount) || maxCount < 1 || maxCount > 100) { + return ctx.text('maxCount must be between 1 and 100', 400); + } + + const nudge = await setNudge({ userId: user.id, sessionId: ctx.req.param('id'), message, everySeconds, maxCount }); + return ctx.json({ nudge }); +}); + +// DELETE /chat/sessions/:id/nudge — disarm. Idempotent: cancelling an absent schedule is success. +chatRouter.delete('/sessions/:id/nudge', async (ctx) => { + const user = ctx.get('user'); + await chatIdentity(user); + await cancelNudge(user.id, ctx.req.param('id')); + return ctx.json({ ok: true }); +}); diff --git a/src/servers/api/chat/nudger.ts b/src/servers/api/chat/nudger.ts new file mode 100644 index 00000000..4d9ae7e5 --- /dev/null +++ b/src/servers/api/chat/nudger.ts @@ -0,0 +1,117 @@ +import { dueNudges, recordNudgeSent, getUserById } from 'officerdb'; +import * as sidecar from '@@/sidecar-registry'; +import { resolveHomeDir } from '@@/user-home'; +import { startClaudeTurn } from './websocket'; + +// Nudger: the schedule that keeps a chat moving while nobody is watching it. +// +// ── Why this can exist at all ── +// +// `startClaudeTurn` already takes `ws: null`. It was extracted that way so `POST /command-hq/commanders/new` +// could drive a turn over HTTP, and that is the whole door this feature needed — a way into a live session +// that is not the chat websocket. Nothing new is exposed; this is a second caller of an existing path. +// +// ── The rule that matters ── +// +// NEVER speak while the agent is working. A prompt delivered mid-turn is noise at best and a derail at +// worst, and the point of this is to restart a stopped agent rather than hurry a busy one. +// +// So `every_seconds` is a FLOOR, not a metronome. When a nudge comes due and the session is generating, +// the row is left alone and reconsidered on the next tick — which means it fires as soon as the turn ends +// rather than sleeping through to the next hour. "At most every N seconds, and as soon as idle after +// that" is what someone asking for this actually wants; a strict hourly alarm would skip an agent that +// finished four minutes late and leave it idle for the other fifty-six. + +const TICK_MS = 60_000; + +/** Is this session mid-turn right now? Asked of the sidecar, immediately before speaking. */ +async function isGenerating(userId: number, sessionId: string): Promise { + try { + const live = await sidecar.listLiveClaudeSessions(userId); + // Matched on BOTH ids on purpose. `sessionKey` is officer's handle for the conversation; + // `claudeSessionId` is the transcript's own filename, and only the agent holds the mapping. A schedule + // stored against one would silently never match a live row keyed by the other. + return live.some((s) => (s.sessionKey === sessionId || s.claudeSessionId === sessionId) && s.isGenerating); + } catch { + // Unknown is not permission. A sidecar that cannot answer might be running a turn, and interrupting one + // is the failure this exists to avoid — so silence is the safe answer. + return true; + } +} + +async function fire(nudge: { id: number; userId: number; sessionId: string; message: string }): Promise { + const user = await getUserById(nudge.userId); + const resolved = await resolveHomeDir(nudge.userId); + if (!user || !resolved.ok) return; + + await new Promise((resolve) => { + let settled = false; + const done = () => { + if (!settled) { + settled = true; + resolve(); + } + }; + // A turn that never reports back must not wedge the tick — the next one will reconsider the row. + const timer = setTimeout(done, 30_000); + + void startClaudeTurn({ + email: user.email, + username: user.username ?? user.email, + userId: nudge.userId, + // No socket. Whoever is watching the chat gets the turn through their own connection, the same way + // they would see a turn started from another tab. + ws: null, + sessionId: nudge.sessionId, + model: 'claude-code/sonnet', + msg: { prompt: nudge.message, resumeSessionId: nudge.sessionId }, + effectivePrompt: nudge.message, + onOutcome: (outcome) => { + clearTimeout(timer); + if (!outcome.ok) console.error(`[nudger] session ${nudge.sessionId}: ${outcome.error}`); + done(); + }, + }).catch((err) => { + clearTimeout(timer); + console.error(`[nudger] session ${nudge.sessionId} threw:`, err instanceof Error ? err.message : err); + done(); + }); + }); + + // Counted whether or not the turn reported success. A nudge that failed to start still consumed its slot; + // retrying it immediately would be a tight loop against whatever refused it. + await recordNudgeSent(nudge.id); +} + +let timer: ReturnType | null = null; + +/** + * Start the tick. Called once from `server.tsx`. + * + * One tick at a time — `running` guards against a slow tick overlapping the next, which would double-fire + * a due nudge. `unref` so this never holds the process open on its own. + */ +export function startNudger(): void { + if (timer) return; + let running = false; + + timer = setInterval(async () => { + if (running) return; + running = true; + try { + for (const nudge of await dueNudges()) { + if (await isGenerating(nudge.userId, nudge.sessionId)) continue; // reconsidered next tick + console.log(`[nudger] session ${nudge.sessionId} idle — nudging (${nudge.sentCount + 1}/${nudge.maxCount})`); + await fire(nudge); + } + } catch (err) { + // A bad tick must not stop every later one. + console.error('[nudger] tick failed:', err instanceof Error ? err.message : err); + } finally { + running = false; + } + }, TICK_MS); + + timer.unref?.(); + console.log('[nudger] armed — checking every 60s'); +}