From 97d933bf6c9e7fae44cdb0bd3a60e8a979cfaf73 Mon Sep 17 00:00:00 2001 From: Seth Karten <32787133+sethkarten@users.noreply.github.com> Date: Thu, 9 Jul 2026 23:17:30 -0700 Subject: [PATCH] Steer heartbeats by default with configurable delivery mode (#358) * feat(coding-agent): steer heartbeats by default with configurable delivery mode Scheduled heartbeat prompts now steer (interrupt the current turn) by default instead of queueing as follow-up messages. Delivery mode is configurable as "steer" or "follow_up" for both user /heartbeat jobs and internal rlm_heartbeat jobs, which share the cron scheduler plumbing. - Add AgentHeartbeatDeliveryMode with DEFAULT_HEARTBEAT_DELIVERY_MODE=steer - Persist deliveryMode on heartbeat/rlm_heartbeat cron jobs - /heartbeat accepts --steer/--follow-up (and --deliver ) - rlm_heartbeat skill create/update accept delivery_mode - shouldDeferHeartbeatCronJob is delivery-mode-aware: steer heartbeats interrupt a plain streaming turn, follow_up still defers - Update CLI/slash help, autocomplete, protocol/connection types, and docs * fix(coding-agent): validate heartbeat delivery mode edge cases Address Macroscope findings on PR #358: - Parse delivery flags that appear after the /heartbeat instruction. - Normalize and validate daemon heartbeat_set deliveryMode before persisting. - Clear stale queued RLM heartbeat follow-ups when only deliveryMode changes. Adds focused regressions for each case. * fix(coding-agent): keep middle heartbeat delivery flags literal Consume /heartbeat delivery flags only from option positions: leading flags, trailing flags, or immediately after the interval. This preserves literal mentions of --follow-up/--steer in the middle of free-form instructions while still supporting trailing opt-out syntax. * fix(coding-agent): clear queued steered heartbeat prompts Address Cursor Bugbot's PR #358 finding: steer-delivered heartbeat prompts can queue in the steering queue, so pause/clear/replace cleanup must remove them by heartbeat queue key just like follow-up heartbeat prompts. Adds queueKey support for steering messages and a regression covering queued steered heartbeat removal. * fix(coding-agent): reject invalid heartbeat delivery flag values Strictly validate --deliver values in /heartbeat command option positions so invalid delivery modes cannot silently become part of the instruction text. Keeps delivery-like text in the middle of free-form instructions literal. * fix(coding-agent): preserve steering queue keys across restart Steered heartbeat prompts rely on their heartbeat queue key for pause/clear and replace cleanup. Preserve that key in update-restart snapshots and restore it when queued steering messages are replayed. * fix(coding-agent): address heartbeat review findings (fixes #358) * fix(coding-agent): preserve heartbeat delivery mode (fixes #358) * fix(coding-agent): preserve queue keys on expanded steer commands Forward queueKey and agentMessageId through AgentSession.steer() for daemon steer commands that use the normal prompt-expansion path. This keeps keyed cleanup working for steered heartbeat prompts regardless of restore path. --------- Co-authored-by: Kevin Thomas --- packages/coding-agent/CHANGELOG.md | 1 + .../skills/rlm-heartbeat/SKILL.md | 23 +- .../src/rlm_heartbeat/__init__.py | 32 +- .../coding-agent/src/core/agent-session.ts | 51 ++- packages/coding-agent/src/core/cron-jobs.ts | 177 ++++++++- .../coding-agent/src/core/slash-commands.ts | 5 +- .../daemon-agent-connection.ts | 9 +- .../in-process-agent-connection.ts | 8 +- .../src/modes/agent-connection/types.ts | 8 +- .../src/modes/daemon/daemon-mode.ts | 60 ++- .../src/modes/daemon/daemon-protocol.ts | 12 +- .../src/modes/interactive/interactive-mode.ts | 23 +- packages/coding-agent/test/cron-jobs.test.ts | 211 +++++++++- .../coding-agent/test/daemon-mode.test.ts | 361 +++++++++++++++++- ...interactive-mode-heartbeat-command.test.ts | 16 +- .../test/kernel-rlm-heartbeat-skill.test.ts | 32 +- .../coding-agent/test/slash-commands.test.ts | 5 +- .../4257-update-restart-resume.test.ts | 23 +- .../4482-heartbeat-injected-prompt.test.ts | 49 +++ .../4519-heartbeat-rebirth.test.ts | 2 +- 20 files changed, 1027 insertions(+), 81 deletions(-) diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index deef57a9b..3d172e3fd 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -3,6 +3,7 @@ ## [Unreleased] - Added `/btw` and `/side` for one-turn inline side questions that use the current context without changing the main session ([ENG-4509](https://linear.app/primeintellect/issue/ENG-4509/add-btw-and-side-side-question-flows)). +- Changed scheduled heartbeat prompts to steer (interrupt the current turn) by default, with a `steer`/`follow_up` delivery mode selectable via `/heartbeat --steer|--follow-up` and the `rlm_heartbeat` skill's `delivery_mode` argument. - Changed the new-chat splash to show only version, model, and cwd metadata and rotate among five example prompts. - Fixed stale heartbeat jobs reopening archived, deleted, or concurrently terminated sessions ([ENG-4519](https://linear.app/primeintellect/issue/ENG-4519/heartbeats-rebirth-sessions-that-were-previously-killed)). diff --git a/packages/coding-agent/skills/rlm-heartbeat/SKILL.md b/packages/coding-agent/skills/rlm-heartbeat/SKILL.md index d7a6dc23d..6ee2d995d 100644 --- a/packages/coding-agent/skills/rlm-heartbeat/SKILL.md +++ b/packages/coding-agent/skills/rlm-heartbeat/SKILL.md @@ -13,6 +13,7 @@ Call directly from IPython: ```python await rlm_heartbeat.create("check test progress", interval="5m", label="tests") +await rlm_heartbeat.create("watch build", delivery_mode="follow_up") await rlm_heartbeat.list() await rlm_heartbeat.update("job-id", status="pause") await rlm_heartbeat.delete("job-id") @@ -22,15 +23,25 @@ await rlm_heartbeat.delete("job-id") - `await rlm_heartbeat.list(include_inactive=False)` — list this session's internal RLM heartbeats. By default this includes active and paused entries. -- `await rlm_heartbeat.create(instruction, interval=None, label=None)` — create - a recurring heartbeat for this session. The default interval is every 5 - minutes. Multiple RLM heartbeats may run at once; use labels to distinguish - them. +- `await rlm_heartbeat.create(instruction, interval=None, label=None, + delivery_mode=None)` — create a recurring heartbeat for this session. The + default interval is every 5 minutes. Multiple RLM heartbeats may run at once; + use labels to distinguish them. `delivery_mode` is `"steer"` (default) or + `"follow_up"`. - `await rlm_heartbeat.update(id, instruction=None, interval=None, label=None, - status=None)` — update one RLM heartbeat by id. `status` may be `"pause"` or - `"resume"`. + status=None, delivery_mode=None)` — update one RLM heartbeat by id. `status` + may be `"pause"` or `"resume"`; `delivery_mode` may be `"steer"` or + `"follow_up"`. - `await rlm_heartbeat.delete(id)` — cancel one RLM heartbeat by id. +## Delivery mode + +Each heartbeat has a delivery mode controlling how the scheduled prompt reaches +the session when it is busy: + +- `steer` (default): interrupt the current turn so the heartbeat runs promptly. +- `follow_up`: wait for the current turn to finish before running the heartbeat. + ## Rules - Use this only for agent-internal recurring checks and long-running task diff --git a/packages/coding-agent/skills/rlm-heartbeat/src/rlm_heartbeat/__init__.py b/packages/coding-agent/skills/rlm-heartbeat/src/rlm_heartbeat/__init__.py index 0f946a7d6..1f57d5f7b 100644 --- a/packages/coding-agent/skills/rlm-heartbeat/src/rlm_heartbeat/__init__.py +++ b/packages/coding-agent/skills/rlm-heartbeat/src/rlm_heartbeat/__init__.py @@ -12,6 +12,17 @@ from typing import Any, Literal from rlm import host_request StatusUpdate = Literal["pause", "resume"] +DeliveryMode = Literal["steer", "follow_up"] + + +def _normalize_delivery_mode(delivery_mode: DeliveryMode | None) -> str | None: + if delivery_mode is None: + return None + if not isinstance(delivery_mode, str): + raise TypeError(f"delivery_mode must be str or None, got {type(delivery_mode).__name__}") + if delivery_mode not in {"steer", "follow_up"}: + raise ValueError('delivery_mode must be "steer", "follow_up", or None') + return delivery_mode async def list(include_inactive: bool = False) -> dict[str, Any]: @@ -21,8 +32,18 @@ async def list(include_inactive: bool = False) -> dict[str, Any]: return await host_request("rlm_heartbeat.list", {"include_inactive": include_inactive}) -async def create(instruction: str, interval: str | None = None, label: str | None = None) -> dict[str, Any]: - """Create an internal recurring heartbeat for the current agent session.""" +async def create( + instruction: str, + interval: str | None = None, + label: str | None = None, + delivery_mode: DeliveryMode | None = None, +) -> dict[str, Any]: + """Create an internal recurring heartbeat for the current agent session. + + delivery_mode controls how the scheduled prompt is delivered when the + session is busy: "steer" (default) interrupts the current turn, "follow_up" + waits for it to finish. + """ if not isinstance(instruction, str): raise TypeError(f"instruction must be str, got {type(instruction).__name__}") payload: dict[str, Any] = {"instruction": instruction} @@ -34,6 +55,9 @@ async def create(instruction: str, interval: str | None = None, label: str | Non if not isinstance(label, str): raise TypeError(f"label must be str or None, got {type(label).__name__}") payload["label"] = label + normalized_delivery_mode = _normalize_delivery_mode(delivery_mode) + if normalized_delivery_mode is not None: + payload["delivery_mode"] = normalized_delivery_mode return await host_request("rlm_heartbeat.create", payload) @@ -43,6 +67,7 @@ async def update( interval: str | None = None, label: str | None = None, status: StatusUpdate | None = None, + delivery_mode: DeliveryMode | None = None, ) -> dict[str, Any]: """Update one internal RLM heartbeat for the current agent session.""" if not isinstance(id, str): @@ -64,6 +89,9 @@ async def update( if status not in {"pause", "resume"}: raise ValueError('status must be "pause", "resume", or None') payload["status"] = status + normalized_delivery_mode = _normalize_delivery_mode(delivery_mode) + if normalized_delivery_mode is not None: + payload["delivery_mode"] = normalized_delivery_mode return await host_request("rlm_heartbeat.update", payload) diff --git a/packages/coding-agent/src/core/agent-session.ts b/packages/coding-agent/src/core/agent-session.ts index bb652242c..014dd2fab 100644 --- a/packages/coding-agent/src/core/agent-session.ts +++ b/packages/coding-agent/src/core/agent-session.ts @@ -93,6 +93,7 @@ import { loadContextTreeChildrenFromDisk, } from "./context-tree.js"; import type { AgentCronJob, AgentRlmHeartbeatController, AgentRlmHeartbeatStatusUpdate } from "./cron-jobs.js"; +import { normalizeHeartbeatDeliveryMode } from "./cron-jobs.js"; import { DEFAULT_THINKING_LEVEL } from "./defaults.js"; import { exportSessionToHtml, type ToolHtmlRenderer } from "./export-html/index.js"; import { createToolHtmlRenderer } from "./export-html/tool-renderer.js"; @@ -416,6 +417,7 @@ type QueuedAgentMessage = UserMessage | CustomMessage; interface QueuedSteeringMessage { text: string; previewLabel?: string; + queueKey?: string; agentMessageId?: string; message: QueuedAgentMessage; } @@ -1477,12 +1479,14 @@ export class AgentSession { if (payload.label !== undefined && typeof payload.label !== "string") { throw new Error("rlm_heartbeat.create label must be a string when provided"); } + const deliveryMode = normalizeHeartbeatDeliveryMode(payload.delivery_mode ?? payload.deliveryMode); return { heartbeat: rlmHeartbeatHostResponse( controller.createRlmHeartbeat({ instruction: payload.instruction, interval: payload.interval, label: payload.label, + deliveryMode, }), ), }; @@ -1503,11 +1507,14 @@ export class AgentSession { if (payload.status !== undefined && !isRlmHeartbeatStatusUpdate(payload.status)) { throw new Error('rlm_heartbeat.update status must be "pause" or "resume" when provided'); } + const rawDeliveryMode = payload.delivery_mode ?? payload.deliveryMode; + const deliveryMode = normalizeHeartbeatDeliveryMode(rawDeliveryMode); if ( payload.instruction === undefined && payload.interval === undefined && payload.label === undefined && - payload.status === undefined + payload.status === undefined && + rawDeliveryMode === undefined ) { throw new Error("rlm_heartbeat.update requires at least one field to update"); } @@ -1517,6 +1524,7 @@ export class AgentSession { interval: payload.interval, label: payload.label, status: payload.status, + deliveryMode, }); return { heartbeat: heartbeat ? rlmHeartbeatHostResponse(heartbeat) : null }; } @@ -3032,7 +3040,11 @@ export class AgentSession { * @param images Optional image attachments to include with the message * @throws Error if text is an extension command */ - async steer(text: string, images?: ImageContent[]): Promise { + async steer( + text: string, + images?: ImageContent[], + options: { queueKey?: string; agentMessageId?: string } = {}, + ): Promise { // Check for extension commands (cannot be queued) if (text.startsWith("/")) { this._throwIfExtensionCommand(text); @@ -3042,7 +3054,10 @@ export class AgentSession { let expandedText = this._expandSkillCommand(text); expandedText = expandPromptTemplate(expandedText, [...this.promptTemplates]); - await this._queueSteer(expandedText, images); + await this._queueSteer(expandedText, images, { + queueKey: options.queueKey, + agentMessageId: options.agentMessageId, + }); } /** @@ -3076,12 +3091,14 @@ export class AgentSession { text: string, images?: ImageContent[], options: { + queueKey?: string; agentMessageId?: string; content?: (TextContent | ImageContent)[]; customMessage?: CustomMessage; } = {}, ): Promise { await this._queueSteer(text, images, { + queueKey: options.queueKey, agentMessageId: options.agentMessageId, content: options.content, message: options.customMessage, @@ -3143,7 +3160,11 @@ export class AgentSession { } return queued; } - await this._queueSteer(text, undefined, { agentMessageId: options.agentMessageId, content }); + await this._queueSteer(text, undefined, { + agentMessageId: options.agentMessageId, + queueKey: options.queueKey, + content, + }); return true; } catch (error) { this._pendingNextTurnMessages.unshift(...pendingNextTurnMessages); @@ -3175,7 +3196,11 @@ export class AgentSession { } return queued; } - await this._queueSteer(text, undefined, { message: queuedMessage, previewLabel: options.previewLabel }); + await this._queueSteer(text, undefined, { + message: queuedMessage, + previewLabel: options.previewLabel, + queueKey: options.queueKey, + }); return true; } catch (error) { this._pendingNextTurnMessages.unshift(...pendingNextTurnMessages); @@ -3191,6 +3216,7 @@ export class AgentSession { images?: ImageContent[], options: { agentMessageId?: string; + queueKey?: string; content?: (TextContent | ImageContent)[]; message?: QueuedAgentMessage; previewLabel?: string; @@ -3207,6 +3233,7 @@ export class AgentSession { this._steeringMessages.push({ text, previewLabel: options.previewLabel, + queueKey: options.queueKey, agentMessageId: options.agentMessageId, message, }); @@ -3492,13 +3519,18 @@ export class AgentSession { } removeQueuedFollowUp(queueKey: string): boolean { - const removed = this._followUpMessages.filter((message) => message.queueKey === queueKey); - if (removed.length === 0) { + const removedSteering = this._steeringMessages.filter((message) => message.queueKey === queueKey); + const removedFollowUp = this._followUpMessages.filter((message) => message.queueKey === queueKey); + if (removedSteering.length === 0 && removedFollowUp.length === 0) { return false; } + this._steeringMessages = this._steeringMessages.filter((message) => message.queueKey !== queueKey); this._followUpMessages = this._followUpMessages.filter((message) => message.queueKey !== queueKey); - const removedMessages = new Set(removed.map((message) => message.message)); - for (const message of removed) { + const removedMessages = new Set([ + ...removedSteering.map((message) => message.message), + ...removedFollowUp.map((message) => message.message), + ]); + for (const message of [...removedSteering, ...removedFollowUp]) { this._rejectAgentMessageDelivery( message.agentMessageId, new Error("Queued agent message was cleared before delivery."), @@ -6990,6 +7022,7 @@ function rlmHeartbeatHostResponse(job: AgentCronJob): Record { id: job.id, status: job.status, label: job.label ?? null, + delivery_mode: job.deliveryMode ?? "steer", instruction: job.prompt, schedule: job.schedule, created_at: job.createdAt, diff --git a/packages/coding-agent/src/core/cron-jobs.ts b/packages/coding-agent/src/core/cron-jobs.ts index f388b7562..50c7a34bd 100644 --- a/packages/coding-agent/src/core/cron-jobs.ts +++ b/packages/coding-agent/src/core/cron-jobs.ts @@ -10,6 +10,11 @@ export type AgentCronJobSource = "cron" | "heartbeat" | "rlm_heartbeat"; export type AgentCronJobRuntimeKind = "top-level" | "subagent"; export type AgentHeartbeatUpdateAction = "pause" | "resume" | "clear"; export type AgentRlmHeartbeatStatusUpdate = "pause" | "resume"; +/** + * How a scheduled heartbeat prompt is delivered when the target session is busy: + * "steer" interrupts the current turn, "follow_up" waits for it to finish. + */ +export type AgentHeartbeatDeliveryMode = "steer" | "follow_up"; export interface AgentCronSchedule { kind: AgentCronScheduleKind; @@ -22,6 +27,8 @@ export interface AgentCronJob { status: AgentCronJobStatus; source?: AgentCronJobSource; runtimeKind?: AgentCronJobRuntimeKind; + /** Delivery mode for heartbeat/rlm_heartbeat jobs when the session is busy. Defaults to "steer". */ + deliveryMode?: AgentHeartbeatDeliveryMode; activeSessionId: string; sessionId: string; sessionFile: string; @@ -48,6 +55,7 @@ export interface CreateAgentCronJobInput { scheduleText: string; source?: AgentCronJobSource; runtimeKind?: AgentCronJobRuntimeKind; + deliveryMode?: AgentHeartbeatDeliveryMode; now?: Date; } @@ -76,13 +84,14 @@ const MAX_TIMEOUT_MS = 2_147_483_647; const ONE_SECOND_MS = 1000; const ONE_MINUTE_MS = 60_000; export const DEFAULT_HEARTBEAT_SCHEDULE = "every 5m"; +export const DEFAULT_HEARTBEAT_DELIVERY_MODE: AgentHeartbeatDeliveryMode = "steer"; export type ParsedHeartbeatCommand = | { type: "status" } | { type: "pause" } | { type: "resume" } | { type: "clear" } - | { type: "set"; schedule: string; instruction: string }; + | { type: "set"; schedule: string; instruction: string; deliveryMode?: AgentHeartbeatDeliveryMode }; export interface AgentCronToolController { getHeartbeat(): AgentCronJob | undefined; @@ -90,13 +99,19 @@ export interface AgentCronToolController { export interface AgentRlmHeartbeatController { listRlmHeartbeats(options?: { includeInactive?: boolean }): AgentCronJob[]; - createRlmHeartbeat(input: { instruction: string; interval?: string; label?: string }): AgentCronJob; + createRlmHeartbeat(input: { + instruction: string; + interval?: string; + label?: string; + deliveryMode?: AgentHeartbeatDeliveryMode; + }): AgentCronJob; updateRlmHeartbeat(input: { id: string; instruction?: string; interval?: string; label?: string; status?: AgentRlmHeartbeatStatusUpdate; + deliveryMode?: AgentHeartbeatDeliveryMode; }): AgentCronJob | undefined; deleteRlmHeartbeat(id: string): AgentCronJob | undefined; } @@ -224,6 +239,7 @@ export class AgentCronJobStore { status: "active", source: "heartbeat", runtimeKind: input.runtimeKind, + deliveryMode: input.deliveryMode ?? DEFAULT_HEARTBEAT_DELIVERY_MODE, activeSessionId: input.activeSessionId, sessionId: input.sessionId, sessionFile: input.sessionFile, @@ -270,6 +286,7 @@ export class AgentCronJobStore { status: "active", source: "rlm_heartbeat", runtimeKind: input.runtimeKind, + deliveryMode: input.deliveryMode ?? DEFAULT_HEARTBEAT_DELIVERY_MODE, activeSessionId: input.activeSessionId, sessionId: input.sessionId, sessionFile: input.sessionFile, @@ -294,6 +311,7 @@ export class AgentCronJobStore { prompt?: string; scheduleText?: string; status?: AgentRlmHeartbeatStatusUpdate; + deliveryMode?: AgentHeartbeatDeliveryMode; now?: Date; }, ): AgentCronJob | undefined { @@ -312,6 +330,9 @@ export class AgentCronJobStore { if (update.label !== undefined) { nextJob = { ...nextJob, label: normalizeOptionalLabel(update.label) }; } + if (update.deliveryMode !== undefined) { + nextJob = { ...nextJob, deliveryMode: update.deliveryMode }; + } if (update.prompt !== undefined) { const prompt = update.prompt.trim(); if (!prompt) { @@ -744,6 +765,22 @@ export function normalizeHeartbeatSchedule(input: string | undefined): string { return text; } +export function normalizeHeartbeatDeliveryMode(value: unknown): AgentHeartbeatDeliveryMode | undefined { + if (value === undefined || value === null) { + return undefined; + } + if (value === "steer" || value === "follow_up") { + return value; + } + throw new Error('Heartbeat delivery mode must be "steer" or "follow_up"'); +} + +export function resolveHeartbeatStreamingBehavior( + deliveryMode: AgentHeartbeatDeliveryMode | undefined, +): "steer" | "followUp" { + return (deliveryMode ?? DEFAULT_HEARTBEAT_DELIVERY_MODE) === "follow_up" ? "followUp" : "steer"; +} + export function parseHeartbeatCommand(input: string): ParsedHeartbeatCommand { const text = input.replace(/^\/heartbeat\b/, "").trim(); if (!text || text === "status") { @@ -759,31 +796,50 @@ export function parseHeartbeatCommand(input: string): ParsedHeartbeatCommand { return { type: "clear" }; } - const option = consumeEveryOption(text); + const leadingDelivery = consumeDeliveryOption(text); + let deliveryMode = leadingDelivery.deliveryMode; + let remaining = leadingDelivery.rest; + + const option = consumeEveryOption(remaining); if (option) { - if (!option.rest) { - throw new Error("Usage: /heartbeat [--every ] "); + const trailingDelivery = consumeDeliveryOption(option.rest); + deliveryMode = trailingDelivery.deliveryMode ?? deliveryMode; + if (!trailingDelivery.rest) { + throw new Error("Usage: /heartbeat [--every ] [--steer|--follow-up] "); } return { type: "set", schedule: normalizeHeartbeatSchedule(option.interval), - instruction: option.rest, + instruction: trailingDelivery.rest, + ...(deliveryMode ? { deliveryMode } : {}), }; } - const leadingSchedule = consumeLeadingEverySchedule(text); + const leadingSchedule = consumeLeadingEverySchedule(remaining); if (leadingSchedule) { - if (!leadingSchedule.rest) { - throw new Error("Usage: /heartbeat [--every ] "); + const trailingDelivery = consumeDeliveryOption(leadingSchedule.rest); + deliveryMode = trailingDelivery.deliveryMode ?? deliveryMode; + if (!trailingDelivery.rest) { + throw new Error("Usage: /heartbeat [--every ] [--steer|--follow-up] "); } return { type: "set", schedule: normalizeHeartbeatSchedule(leadingSchedule.interval), - instruction: leadingSchedule.rest, + instruction: trailingDelivery.rest, + ...(deliveryMode ? { deliveryMode } : {}), }; } - return { type: "set", schedule: DEFAULT_HEARTBEAT_SCHEDULE, instruction: text }; + remaining = remaining.trim(); + if (!remaining) { + throw new Error("Usage: /heartbeat [--every ] [--steer|--follow-up] "); + } + return { + type: "set", + schedule: DEFAULT_HEARTBEAT_SCHEDULE, + instruction: remaining, + ...(deliveryMode ? { deliveryMode } : {}), + }; } export function nextRunAtForSchedule(schedule: AgentCronSchedule, after: Date): Date | undefined { @@ -830,6 +886,70 @@ export function createAgentHeartbeatToolDefinitions(controller: AgentCronToolCon ]; } +function consumeDeliveryOption(text: string): { deliveryMode: AgentHeartbeatDeliveryMode | undefined; rest: string } { + let rest = text.trim(); + if (/(?:^|\s)--deliver=?$/i.test(rest)) { + throw new Error('Heartbeat delivery mode must be "steer" or "follow_up"'); + } + let deliveryMode: AgentHeartbeatDeliveryMode | undefined; + let leading = consumeLeadingDeliveryFlag(rest); + while (leading) { + deliveryMode = leading.deliveryMode; + rest = leading.rest.trim(); + leading = consumeLeadingDeliveryFlag(rest); + } + + let trailing = consumeTrailingDeliveryFlag(rest); + let trailingDeliveryMode: AgentHeartbeatDeliveryMode | undefined; + while (trailing) { + // Consume all trailing flags, but keep the rightmost flag's mode because it + // is textually latest and should win over earlier flags. + trailingDeliveryMode ??= trailing.deliveryMode; + rest = trailing.rest.trim(); + trailing = consumeTrailingDeliveryFlag(rest); + } + return { deliveryMode: trailingDeliveryMode ?? deliveryMode, rest }; +} + +function consumeLeadingDeliveryFlag( + text: string, +): { deliveryMode: AgentHeartbeatDeliveryMode; rest: string } | undefined { + const match = /^--(?:deliver(?:=|\s+)(\S+)|(steer)|(follow[-_]up))(?:\s+|$)([\s\S]*)$/i.exec(text); + if (!match) { + return undefined; + } + return { + deliveryMode: parseDeliveryModeToken(match[1] ?? match[2] ?? match[3] ?? ""), + rest: match[4]?.trim() ?? "", + }; +} + +function consumeTrailingDeliveryFlag( + text: string, +): { deliveryMode: AgentHeartbeatDeliveryMode; rest: string } | undefined { + const deliverWithSpace = /^([\s\S]*?)\s+--deliver\s+(\S+)$/i.exec(text); + if (deliverWithSpace) { + return { deliveryMode: parseDeliveryModeToken(deliverWithSpace[2] ?? ""), rest: deliverWithSpace[1] ?? "" }; + } + const deliverWithEquals = /^([\s\S]*?)\s+--deliver=(\S+)$/i.exec(text); + if (deliverWithEquals) { + return { deliveryMode: parseDeliveryModeToken(deliverWithEquals[2] ?? ""), rest: deliverWithEquals[1] ?? "" }; + } + const shorthand = /^([\s\S]*?)\s+--(steer|follow[-_]up)$/i.exec(text); + if (shorthand) { + return { deliveryMode: parseDeliveryModeToken(shorthand[2] ?? ""), rest: shorthand[1] ?? "" }; + } + return undefined; +} + +function parseDeliveryModeToken(token: string): AgentHeartbeatDeliveryMode { + const normalized = token.toLowerCase().replace("-", "_"); + if (normalized === "steer" || normalized === "follow_up") { + return normalized; + } + throw new Error('Heartbeat delivery mode must be "steer" or "follow_up"'); +} + function consumeEveryOption(text: string): { interval: string; rest: string } | undefined { const match = /^--every(?:=|\s+)(?:"([^"]+)"|'([^']+)'|(\d+\s*(?:s|sec|secs|second|seconds|m|min|mins|minute|minutes|h|hr|hrs|hour|hours))|(\S+))(?:\s+|$)([\s\S]*)$/i.exec( @@ -855,7 +975,8 @@ function consumeLeadingEverySchedule(text: string): { interval: string; rest: st rest: text .slice(match[0].length) .trim() - .replace(/^--\s*/, "") + // Strip only a standalone "--" separator, never a flag like "--follow-up". + .replace(/^--(?=\s|$)/, "") .trim(), }; } @@ -865,15 +986,26 @@ export function isHeartbeatCronJob(job: AgentCronJob): boolean { } export function shouldDeferHeartbeatCronJob(job: AgentCronJob, activity: HeartbeatCronSessionActivity): boolean { - return ( - isHeartbeatCronJob(job) && - (activity.isStreaming || - activity.isCompacting === true || - activity.isRetrying === true || - activity.isBashRunning || - activity.hasAcceptedPromptInFlight === true || - activity.pendingMessageCount > 0) - ); + if (!isHeartbeatCronJob(job)) { + return false; + } + // States where delivering a heartbeat is unsafe or would stack redundant work, + // regardless of delivery mode. + const busyBesidesStreaming = + activity.isCompacting === true || + activity.isRetrying === true || + activity.isBashRunning || + activity.hasAcceptedPromptInFlight === true || + activity.pendingMessageCount > 0; + if (busyBesidesStreaming) { + return true; + } + // "steer" heartbeats interrupt the current turn, so a plain streaming turn must + // not defer them; "follow_up" heartbeats wait, so streaming still defers. + if (resolveHeartbeatStreamingBehavior(job.deliveryMode) === "steer") { + return false; + } + return activity.isStreaming; } function nextCronRunAfter(expression: string, after: Date): Date { @@ -1070,6 +1202,9 @@ function isAgentCronJob(value: unknown): value is AgentCronJob { (candidate.runtimeKind === undefined || candidate.runtimeKind === "top-level" || candidate.runtimeKind === "subagent") && + (candidate.deliveryMode === undefined || + candidate.deliveryMode === "steer" || + candidate.deliveryMode === "follow_up") && typeof candidate.activeSessionId === "string" && typeof candidate.sessionId === "string" && typeof candidate.sessionFile === "string" && diff --git a/packages/coding-agent/src/core/slash-commands.ts b/packages/coding-agent/src/core/slash-commands.ts index 69b84f5fc..a0d6cdcb4 100644 --- a/packages/coding-agent/src/core/slash-commands.ts +++ b/packages/coding-agent/src/core/slash-commands.ts @@ -89,8 +89,9 @@ const CANONICAL_BUILTIN_SLASH_COMMANDS: ReadonlyArray = [ }, { name: "heartbeat", - description: "Set or view a persistent heartbeat; supports pause, resume, stop, and clear", - argumentHint: "[status|pause|resume|stop|every ]", + description: + "Set or view a persistent heartbeat; delivery defaults to steer, use --follow-up to queue; supports pause, resume, stop, and clear", + argumentHint: "[status|pause|resume|stop|[every ] [--steer|--follow-up] ]", takesArgument: true, }, { name: "resume", description: "Resume a different session" }, diff --git a/packages/coding-agent/src/modes/agent-connection/daemon-agent-connection.ts b/packages/coding-agent/src/modes/agent-connection/daemon-agent-connection.ts index c471da5b2..b1ec3f9a4 100644 --- a/packages/coding-agent/src/modes/agent-connection/daemon-agent-connection.ts +++ b/packages/coding-agent/src/modes/agent-connection/daemon-agent-connection.ts @@ -3,7 +3,7 @@ import type { AgentMessage, ThinkingLevel } from "@earendil-works/pi-agent-core" import type { ImageContent, Transport } from "@earendil-works/pi-ai"; import type { CompactionResult } from "../../core/compaction/index.js"; import type { ContextTreeNode } from "../../core/context-tree.js"; -import type { AgentCronJob, AgentHeartbeatUpdateAction } from "../../core/cron-jobs.js"; +import type { AgentCronJob, AgentHeartbeatDeliveryMode, AgentHeartbeatUpdateAction } from "../../core/cron-jobs.js"; import type { RefinementResult } from "../../core/refinement/index.js"; import type { DeleteSessionFileResult } from "../../core/session-file-actions.js"; import type { SessionStats } from "../../core/session-stats.js"; @@ -343,12 +343,17 @@ export class DaemonAgentConnection implements AgentConnection { return data.heartbeat ?? undefined; } - async setHeartbeat(schedule: string, instruction: string): Promise { + async setHeartbeat( + schedule: string, + instruction: string, + deliveryMode?: AgentHeartbeatDeliveryMode, + ): Promise { const data = await this.requestData<{ heartbeat: AgentCronJob }>({ type: "heartbeat_set", activeSessionId: this.activeSessionId, schedule, prompt: instruction, + ...(deliveryMode ? { deliveryMode } : {}), }); return data.heartbeat; } diff --git a/packages/coding-agent/src/modes/agent-connection/in-process-agent-connection.ts b/packages/coding-agent/src/modes/agent-connection/in-process-agent-connection.ts index 7649d6cb9..0c2aa61d5 100644 --- a/packages/coding-agent/src/modes/agent-connection/in-process-agent-connection.ts +++ b/packages/coding-agent/src/modes/agent-connection/in-process-agent-connection.ts @@ -4,7 +4,7 @@ import type { ImageContent, Transport } from "@earendil-works/pi-ai"; import type { AgentSessionRuntime } from "../../core/agent-session-runtime.js"; import type { CompactionResult } from "../../core/compaction/index.js"; import type { ContextTreeNode } from "../../core/context-tree.js"; -import type { AgentCronJob, AgentHeartbeatUpdateAction } from "../../core/cron-jobs.js"; +import type { AgentCronJob, AgentHeartbeatDeliveryMode, AgentHeartbeatUpdateAction } from "../../core/cron-jobs.js"; import type { RefinementResult } from "../../core/refinement/index.js"; import { type DeleteSessionFileResult, deleteSessionFile } from "../../core/session-file-actions.js"; import { SessionManager } from "../../core/session-manager.js"; @@ -178,7 +178,11 @@ export class InProcessAgentConnection implements AgentConnection { return undefined; } - async setHeartbeat(_schedule: string, _instruction: string): Promise { + async setHeartbeat( + _schedule: string, + _instruction: string, + _deliveryMode?: AgentHeartbeatDeliveryMode, + ): Promise { throw new Error("Heartbeats require daemon mode"); } diff --git a/packages/coding-agent/src/modes/agent-connection/types.ts b/packages/coding-agent/src/modes/agent-connection/types.ts index 2af19911f..69b4f4e20 100644 --- a/packages/coding-agent/src/modes/agent-connection/types.ts +++ b/packages/coding-agent/src/modes/agent-connection/types.ts @@ -3,7 +3,7 @@ import type { Api, ImageContent, Model, TextContent, Transport, Usage } from "@e import type { AuthSourceToken } from "../../core/auth-storage.js"; import type { CompactionResult } from "../../core/compaction/index.js"; import type { ContextTreeNode } from "../../core/context-tree.js"; -import type { AgentCronJob, AgentHeartbeatUpdateAction } from "../../core/cron-jobs.js"; +import type { AgentCronJob, AgentHeartbeatDeliveryMode, AgentHeartbeatUpdateAction } from "../../core/cron-jobs.js"; import type { ReplayBuiltInToolName } from "../../core/extensions/index.js"; import type { GoalState } from "../../core/goals.js"; import type { RefinementResult } from "../../core/refinement/index.js"; @@ -541,7 +541,11 @@ export interface AgentConnection { addCronJob(schedule: string, prompt: string): Promise; cancelCronJob(jobId: string): Promise; getHeartbeat(): Promise; - setHeartbeat(schedule: string, instruction: string): Promise; + setHeartbeat( + schedule: string, + instruction: string, + deliveryMode?: AgentHeartbeatDeliveryMode, + ): Promise; updateHeartbeat(action: AgentHeartbeatUpdateAction): Promise; getUserMessagesForForking(): Promise; getLastAssistantText(): Promise; diff --git a/packages/coding-agent/src/modes/daemon/daemon-mode.ts b/packages/coding-agent/src/modes/daemon/daemon-mode.ts index 5a10efbbc..a28aa0f22 100644 --- a/packages/coding-agent/src/modes/daemon/daemon-mode.ts +++ b/packages/coding-agent/src/modes/daemon/daemon-mode.ts @@ -64,11 +64,14 @@ import { type AgentCronJob, AgentCronJobStore, AgentCronScheduler, + type AgentHeartbeatDeliveryMode, type AgentHeartbeatUpdateAction, createAgentHeartbeatToolDefinitions, DEFAULT_HEARTBEAT_SCHEDULE, isHeartbeatCronJob, + normalizeHeartbeatDeliveryMode, normalizeHeartbeatSchedule, + resolveHeartbeatStreamingBehavior, shouldDeferHeartbeatCronJob, } from "../../core/cron-jobs.js"; import type { @@ -697,11 +700,12 @@ export class AgentDaemon { return current && this.isCronJobRunnableForState(current, state, requirePersistedJob) ? current : undefined; }; if (isHeartbeatCronJob(runnableJob)) { + const streamingBehavior = resolveHeartbeatStreamingBehavior(runnableJob.deliveryMode); const didPrompt = await this.promptHeartbeatWithAgentMessagePreparingGuard( state, runnableJob, { - streamingBehavior: "followUp", + streamingBehavior, followUpQueueKey: `heartbeat:${runnableJob.id}`, source: "rpc", }, @@ -734,7 +738,7 @@ export class AgentDaemon { ): Promise { return this.withAgentMessagePreparingGuard( state, - (session) => { + (session, clearPreparing) => { const prompt = options?.agentMessageId === undefined ? session.prompt.bind(session) @@ -742,6 +746,9 @@ export class AgentDaemon { return prompt(message, { ...options, preflightResult: (didSucceed) => { + if (didSucceed && session.isStreaming) { + clearPreparing(); + } options?.preflightResult?.(didSucceed); }, }); @@ -759,10 +766,13 @@ export class AgentDaemon { let promptJob = job; return this.withAgentMessagePreparingGuard( state, - (session) => + (session, clearPreparing) => session.promptHeartbeat(promptJob, { ...options, preflightResult: (didSucceed) => { + if (didSucceed && session.isStreaming) { + clearPreparing(); + } options?.preflightResult?.(didSucceed); }, }), @@ -782,7 +792,7 @@ export class AgentDaemon { private async withAgentMessagePreparingGuard( state: ActiveSessionState, - run: (session: AgentSession) => Promise, + run: (session: AgentSession, clearPreparing: () => void) => Promise, canRun?: () => boolean, ): Promise { const activeSessionId = state.activeSessionId; @@ -805,11 +815,11 @@ export class AgentDaemon { }; try { await this.agentMessageTargetLocks.get(activeSessionId)?.catch(() => undefined); - const session = state.runtime.session; if (canRun && !canRun()) { return false; } - await run(session); + const session = state.runtime.session; + await run(session, clearPreparing); return true; } finally { clearPreparing(); @@ -835,7 +845,12 @@ export class AgentDaemon { return job; } - private createHeartbeatForState(state: ActiveSessionState, schedule: string, instruction: string): AgentCronJob { + private createHeartbeatForState( + state: ActiveSessionState, + schedule: string, + instruction: string, + deliveryMode?: AgentHeartbeatDeliveryMode, + ): AgentCronJob { const session = state.runtime.session; const sessionFile = session.sessionFile; if (!sessionFile) { @@ -850,6 +865,7 @@ export class AgentDaemon { runtimeKind: state.runtime.metadata.kind, scheduleText: normalizeHeartbeatSchedule(schedule), prompt: instruction, + deliveryMode: deliveryMode ?? previousHeartbeat?.deliveryMode, }); if (previousHeartbeat) { this.removeQueuedHeartbeatFollowUp(state, previousHeartbeat); @@ -877,7 +893,7 @@ export class AgentDaemon { private createRlmHeartbeatForState( state: ActiveSessionState, - input: { instruction: string; interval?: string; label?: string }, + input: { instruction: string; interval?: string; label?: string; deliveryMode?: AgentHeartbeatDeliveryMode }, ): AgentCronJob { const session = state.runtime.session; const sessionFile = session.sessionFile; @@ -893,6 +909,7 @@ export class AgentDaemon { label: input.label, scheduleText: normalizeHeartbeatSchedule(input.interval ?? DEFAULT_HEARTBEAT_SCHEDULE), prompt: input.instruction, + deliveryMode: input.deliveryMode, }); this.cronScheduler.wake(); return job; @@ -900,16 +917,29 @@ export class AgentDaemon { private updateRlmHeartbeatForState( state: ActiveSessionState, - input: { id: string; instruction?: string; interval?: string; label?: string; status?: "pause" | "resume" }, + input: { + id: string; + instruction?: string; + interval?: string; + label?: string; + status?: "pause" | "resume"; + deliveryMode?: AgentHeartbeatDeliveryMode; + }, ): AgentCronJob | undefined { const job = this.cronStore.updateRlmHeartbeat(state.activeSessionId, input.id, { label: input.label, prompt: input.instruction, scheduleText: input.interval ? normalizeHeartbeatSchedule(input.interval) : undefined, status: input.status, + deliveryMode: input.deliveryMode, }); if (job) { - if (input.instruction !== undefined || input.interval !== undefined || input.status === "pause") { + if ( + input.instruction !== undefined || + input.interval !== undefined || + input.status === "pause" || + input.deliveryMode !== undefined + ) { this.removeQueuedHeartbeatFollowUp(state, job); } this.cronScheduler.wake(); @@ -1625,12 +1655,16 @@ export class AgentDaemon { const state = this.getSessionState(command.activeSessionId); if (command.expandPromptTemplates === false) { await state.runtime.session.restoreSteeringMessage(command.message, command.images, { + queueKey: command.queueKey, agentMessageId: command.agentMessageId, content: command.content, customMessage: command.customMessage, }); } else { - await state.runtime.session.steer(command.message, command.images); + await state.runtime.session.steer(command.message, command.images, { + queueKey: command.queueKey, + agentMessageId: command.agentMessageId, + }); } return success(command.id, "steer"); } @@ -1923,7 +1957,8 @@ export class AgentDaemon { case "heartbeat_set": { const state = this.getSessionState(command.activeSessionId); - const heartbeat = this.createHeartbeatForState(state, command.schedule, command.prompt); + const deliveryMode = normalizeHeartbeatDeliveryMode(command.deliveryMode); + const heartbeat = this.createHeartbeatForState(state, command.schedule, command.prompt, deliveryMode); return success(command.id, "heartbeat_set", { heartbeat }); } @@ -2552,6 +2587,7 @@ export class AgentDaemon { message: message.text, ...(message.content ? { content: message.content } : {}), ...(message.images ? { images: message.images } : {}), + ...(message.queueKey ? { queueKey: message.queueKey } : {}), ...(message.agentMessageId ? { agentMessageId: message.agentMessageId } : {}), ...(message.customMessage ? { customMessage: message.customMessage } : {}), })), diff --git a/packages/coding-agent/src/modes/daemon/daemon-protocol.ts b/packages/coding-agent/src/modes/daemon/daemon-protocol.ts index 8682a8529..72b5eb5b6 100644 --- a/packages/coding-agent/src/modes/daemon/daemon-protocol.ts +++ b/packages/coding-agent/src/modes/daemon/daemon-protocol.ts @@ -7,7 +7,7 @@ import type { } from "../../core/agent-messages.js"; import type { AgentSessionRuntimeConfig } from "../../core/agent-session-config.js"; import type { AgentSessionRuntimeMetadata } from "../../core/agent-session-runtime.js"; -import type { AgentCronJob, AgentHeartbeatUpdateAction } from "../../core/cron-jobs.js"; +import type { AgentCronJob, AgentHeartbeatDeliveryMode, AgentHeartbeatUpdateAction } from "../../core/cron-jobs.js"; import type { CustomMessage } from "../../core/messages.js"; import type { SessionCwdIssue } from "../../core/session-cwd.js"; import type { DeleteSessionFileResult } from "../../core/session-file-actions.js"; @@ -285,6 +285,7 @@ export type DaemonCommand = message: string; content?: (TextContent | ImageContent)[]; images?: ImageContent[]; + queueKey?: string; expandPromptTemplates?: boolean; agentMessageId?: string; customMessage?: CustomMessage; @@ -349,7 +350,14 @@ export type DaemonCommand = | { id?: string; type: "cron_add"; activeSessionId: string; schedule: string; prompt: string } | { id?: string; type: "cron_cancel"; jobId: string } | { id?: string; type: "heartbeat_get"; activeSessionId: string } - | { id?: string; type: "heartbeat_set"; activeSessionId: string; schedule: string; prompt: string } + | { + id?: string; + type: "heartbeat_set"; + activeSessionId: string; + schedule: string; + prompt: string; + deliveryMode?: AgentHeartbeatDeliveryMode; + } | { id?: string; type: "heartbeat_update"; activeSessionId: string; action: AgentHeartbeatUpdateAction } | { id?: string; type: "set_model"; activeSessionId: string; provider: string; modelId: string } | { id?: string; type: "cycle_model"; activeSessionId: string; direction?: "forward" | "backward" } diff --git a/packages/coding-agent/src/modes/interactive/interactive-mode.ts b/packages/coding-agent/src/modes/interactive/interactive-mode.ts index 7d8563af6..550bfabb4 100644 --- a/packages/coding-agent/src/modes/interactive/interactive-mode.ts +++ b/packages/coding-agent/src/modes/interactive/interactive-mode.ts @@ -67,7 +67,7 @@ import { uploadAgentTraceFile, } from "../../core/agent-traces.js"; import { isNoModelsAvailableMessage } from "../../core/auth-guidance.js"; -import { type AgentCronJob, parseHeartbeatCommand } from "../../core/cron-jobs.js"; +import { type AgentCronJob, DEFAULT_HEARTBEAT_DELIVERY_MODE, parseHeartbeatCommand } from "../../core/cron-jobs.js"; import type { AutocompleteProviderFactory, ContextUsage, @@ -425,6 +425,16 @@ const HEARTBEAT_ARGUMENT_COMPLETIONS: AutocompleteItem[] = [ label: "every ", description: "Set an interval, then add an instruction: /heartbeat every 10s Scan the logs", }, + { + value: "--steer ", + label: "--steer ", + description: "Deliver by interrupting the current turn (default)", + }, + { + value: "--follow-up ", + label: "--follow-up ", + description: "Deliver as a follow-up after the current turn finishes", + }, ]; const DEAD_TERMINAL_ERROR_CODES = new Set(["EIO", "EPIPE", "ENOTCONN"]); @@ -8196,9 +8206,15 @@ export class InteractiveMode { return; } case "set": { - const heartbeat = await this.agentConnection.setHeartbeat(command.schedule, command.instruction); + const heartbeat = await this.agentConnection.setHeartbeat( + command.schedule, + command.instruction, + command.deliveryMode, + ); this.patchConnectionState({ heartbeat }); - this.showStatus(`Heartbeat set\nNext run: ${heartbeat.nextRunAt ?? "-"}`); + this.showStatus( + `Heartbeat set\nDelivery: ${heartbeat.deliveryMode ?? DEFAULT_HEARTBEAT_DELIVERY_MODE}\nNext run: ${heartbeat.nextRunAt ?? "-"}`, + ); return; } case "pause": { @@ -8249,6 +8265,7 @@ export class InteractiveMode { "", `${theme.fg("dim", "Status:")} ${job.status}`, `${theme.fg("dim", "Every:")} ${job.schedule.expression}`, + `${theme.fg("dim", "Delivery:")} ${job.deliveryMode ?? DEFAULT_HEARTBEAT_DELIVERY_MODE}`, `${theme.fg("dim", "Instruction:")} ${job.prompt}`, `${theme.fg("dim", "Next:")} ${next}`, `${theme.fg("dim", "Last:")} ${last}`, diff --git a/packages/coding-agent/test/cron-jobs.test.ts b/packages/coding-agent/test/cron-jobs.test.ts index 525e9b741..add488b48 100644 --- a/packages/coding-agent/test/cron-jobs.test.ts +++ b/packages/coding-agent/test/cron-jobs.test.ts @@ -7,8 +7,10 @@ import { AgentCronJobStore, AgentCronScheduler, createAgentHeartbeatToolDefinitions, + normalizeHeartbeatDeliveryMode, parseAgentCronSchedule, parseHeartbeatCommand, + resolveHeartbeatStreamingBehavior, shouldDeferHeartbeatCronJob, } from "../src/core/cron-jobs.js"; @@ -49,7 +51,7 @@ describe("parseHeartbeatCommand", () => { expect(parseHeartbeatCommand("/heartbeat stop")).toEqual({ type: "clear" }); }); - it("defaults new heartbeat instructions to every five minutes", () => { + it("leaves delivery mode unset when an instruction omits the delivery option", () => { expect(parseHeartbeatCommand("/heartbeat check on me")).toEqual({ type: "set", schedule: "every 5m", @@ -74,6 +76,82 @@ describe("parseHeartbeatCommand", () => { instruction: "check status", }); }); + + it("opts into follow-up delivery via --follow-up before or after the interval", () => { + expect(parseHeartbeatCommand("/heartbeat check on me --follow-up")).toEqual({ + type: "set", + schedule: "every 5m", + instruction: "check on me", + deliveryMode: "follow_up", + }); + expect(parseHeartbeatCommand("/heartbeat --follow-up check on me")).toEqual({ + type: "set", + schedule: "every 5m", + instruction: "check on me", + deliveryMode: "follow_up", + }); + expect(parseHeartbeatCommand("/heartbeat --follow-up --every 30s check on me")).toEqual({ + type: "set", + schedule: "every 30s", + instruction: "check on me", + deliveryMode: "follow_up", + }); + expect(parseHeartbeatCommand("/heartbeat every 10m --follow-up check status")).toEqual({ + type: "set", + schedule: "every 10m", + instruction: "check status", + deliveryMode: "follow_up", + }); + expect(parseHeartbeatCommand("/heartbeat --deliver follow_up check status")).toEqual({ + type: "set", + schedule: "every 5m", + instruction: "check status", + deliveryMode: "follow_up", + }); + expect(parseHeartbeatCommand("/heartbeat check status --deliver=follow_up")).toEqual({ + type: "set", + schedule: "every 5m", + instruction: "check status", + deliveryMode: "follow_up", + }); + }); + + it("rejects invalid delivery mode values in delivery flag positions", () => { + for (const command of [ + "/heartbeat --deliver", + "/heartbeat --deliver=", + "/heartbeat check status --deliver", + "/heartbeat check status --deliver=", + ]) { + expect(() => parseHeartbeatCommand(command)).toThrow('Heartbeat delivery mode must be "steer" or "follow_up"'); + } + expect(() => parseHeartbeatCommand("/heartbeat --deliver later check status")).toThrow( + 'Heartbeat delivery mode must be "steer" or "follow_up"', + ); + expect(() => parseHeartbeatCommand("/heartbeat check status --deliver=later")).toThrow( + 'Heartbeat delivery mode must be "steer" or "follow_up"', + ); + expect(() => parseHeartbeatCommand("/heartbeat every 10m --deliver later check status")).toThrow( + 'Heartbeat delivery mode must be "steer" or "follow_up"', + ); + }); + + it("keeps delivery-like flags in the middle of an instruction as instruction text", () => { + expect(parseHeartbeatCommand("/heartbeat remind me to mention --follow-up in docs")).toEqual({ + type: "set", + schedule: "every 5m", + instruction: "remind me to mention --follow-up in docs", + }); + }); + + it("accepts an explicit --steer flag that keeps the default delivery mode", () => { + expect(parseHeartbeatCommand("/heartbeat --steer check on me")).toEqual({ + type: "set", + schedule: "every 5m", + instruction: "check on me", + deliveryMode: "steer", + }); + }); }); describe("AgentCronJobStore", () => { @@ -208,6 +286,54 @@ describe("AgentCronJobStore", () => { expect(store.list().find((job) => job.id === first.id)).toMatchObject({ status: "cancelled" }); }); + it("defaults heartbeats to steer delivery and persists an explicit follow_up opt-out", () => { + const store = new AgentCronJobStore(makeStorePath(tempDirs)); + const steerHeartbeat = store.createHeartbeat({ + activeSessionId: "active-1", + sessionId: "session-1", + sessionFile: "/tmp/session.jsonl", + cwd: "/tmp/project", + scheduleText: "every 5m", + prompt: "steer by default", + now: start, + }); + expect(steerHeartbeat.deliveryMode).toBe("steer"); + + const followUpHeartbeat = store.createHeartbeat({ + activeSessionId: "active-2", + sessionId: "session-2", + sessionFile: "/tmp/session-2.jsonl", + cwd: "/tmp/project", + scheduleText: "every 5m", + prompt: "queue as follow-up", + deliveryMode: "follow_up", + now: start, + }); + expect(followUpHeartbeat.deliveryMode).toBe("follow_up"); + expect(store.getHeartbeat("active-2")).toMatchObject({ deliveryMode: "follow_up" }); + }); + + it("defaults RLM heartbeats to steer delivery and updates the delivery mode", () => { + const store = new AgentCronJobStore(makeStorePath(tempDirs)); + const rlmHeartbeat = store.createRlmHeartbeat({ + activeSessionId: "active-1", + sessionId: "session-1", + sessionFile: "/tmp/session.jsonl", + cwd: "/tmp/project", + scheduleText: "every 5m", + prompt: "watch progress", + now: start, + }); + expect(rlmHeartbeat.deliveryMode).toBe("steer"); + + const updated = store.updateRlmHeartbeat("active-1", rlmHeartbeat.id, { + deliveryMode: "follow_up", + now: new Date("2026-01-01T12:35:00.000Z"), + }); + expect(updated).toMatchObject({ id: rlmHeartbeat.id, deliveryMode: "follow_up" }); + expect(store.listRlmHeartbeats("active-1")[0]).toMatchObject({ deliveryMode: "follow_up" }); + }); + it("pauses, resumes, and clears heartbeat state", () => { const store = new AgentCronJobStore(makeStorePath(tempDirs)); const job = store.createHeartbeat({ @@ -832,9 +958,9 @@ describe("shouldDeferHeartbeatCronJob", () => { runCount: 0, }; - it("defers user and RLM heartbeats while the target session is working", () => { + it("defers follow-up heartbeats while the target session is working", () => { for (const source of ["heartbeat", "rlm_heartbeat"] as const) { - const job = { ...baseJob, source }; + const job = { ...baseJob, source, deliveryMode: "follow_up" as const }; expect( shouldDeferHeartbeatCronJob(job, { @@ -860,6 +986,67 @@ describe("shouldDeferHeartbeatCronJob", () => { } }); + it("lets steer heartbeats interrupt a plain streaming turn", () => { + for (const source of ["heartbeat", "rlm_heartbeat"] as const) { + // Default (undefined) delivery mode is steer. + for (const job of [ + { ...baseJob, source }, + { ...baseJob, source, deliveryMode: "steer" as const }, + ]) { + expect( + shouldDeferHeartbeatCronJob(job, { + isStreaming: true, + isBashRunning: false, + pendingMessageCount: 0, + }), + ).toBe(false); + } + } + }); + + it("defers steer heartbeats when delivering would be unsafe or would stack work", () => { + const job = { ...baseJob, source: "heartbeat" as const, deliveryMode: "steer" as const }; + + expect( + shouldDeferHeartbeatCronJob(job, { + isStreaming: true, + isCompacting: true, + isBashRunning: false, + pendingMessageCount: 0, + }), + ).toBe(true); + expect( + shouldDeferHeartbeatCronJob(job, { + isStreaming: false, + isBashRunning: true, + pendingMessageCount: 0, + }), + ).toBe(true); + expect( + shouldDeferHeartbeatCronJob(job, { + isStreaming: false, + isBashRunning: false, + pendingMessageCount: 1, + }), + ).toBe(true); + expect( + shouldDeferHeartbeatCronJob(job, { + isStreaming: false, + isRetrying: true, + isBashRunning: false, + pendingMessageCount: 0, + }), + ).toBe(true); + expect( + shouldDeferHeartbeatCronJob(job, { + isStreaming: false, + isBashRunning: false, + hasAcceptedPromptInFlight: true, + pendingMessageCount: 0, + }), + ).toBe(true); + }); + it("allows heartbeats when the target session is idle", () => { expect( shouldDeferHeartbeatCronJob( @@ -879,6 +1066,24 @@ describe("shouldDeferHeartbeatCronJob", () => { }); }); +describe("heartbeat delivery mode", () => { + it("normalizes valid delivery modes and rejects invalid ones", () => { + expect(normalizeHeartbeatDeliveryMode(undefined)).toBeUndefined(); + expect(normalizeHeartbeatDeliveryMode(null)).toBeUndefined(); + expect(normalizeHeartbeatDeliveryMode("steer")).toBe("steer"); + expect(normalizeHeartbeatDeliveryMode("follow_up")).toBe("follow_up"); + expect(() => normalizeHeartbeatDeliveryMode("followup")).toThrow( + 'Heartbeat delivery mode must be "steer" or "follow_up"', + ); + }); + + it("resolves delivery mode to a streaming behavior, defaulting to steer", () => { + expect(resolveHeartbeatStreamingBehavior(undefined)).toBe("steer"); + expect(resolveHeartbeatStreamingBehavior("steer")).toBe("steer"); + expect(resolveHeartbeatStreamingBehavior("follow_up")).toBe("followUp"); + }); +}); + describe("createAgentHeartbeatToolDefinitions", () => { it("exposes only read-only user heartbeat inspection to the model", () => { const tools = createAgentHeartbeatToolDefinitions({ diff --git a/packages/coding-agent/test/daemon-mode.test.ts b/packages/coding-agent/test/daemon-mode.test.ts index 5169940c5..5ca64e320 100644 --- a/packages/coding-agent/test/daemon-mode.test.ts +++ b/packages/coding-agent/test/daemon-mode.test.ts @@ -2872,7 +2872,12 @@ describe("daemon mode helpers", () => { ).runCronJob.bind(daemon); const result = await runCronJob( - makeCronJob({ id: "heartbeat-1", source: "heartbeat", activeSessionId: state.activeSessionId }), + makeCronJob({ + id: "heartbeat-1", + source: "heartbeat", + activeSessionId: state.activeSessionId, + deliveryMode: "follow_up", + }), ); expect(result).toBe("skipped"); @@ -2922,10 +2927,20 @@ describe("daemon mode helpers", () => { ).runCronJob.bind(daemon); const first = await runCronJob( - makeCronJob({ id: "rlm-1", source: "rlm_heartbeat", activeSessionId: state.activeSessionId }), + makeCronJob({ + id: "rlm-1", + source: "rlm_heartbeat", + activeSessionId: state.activeSessionId, + deliveryMode: "follow_up", + }), ); const second = await runCronJob( - makeCronJob({ id: "rlm-2", source: "rlm_heartbeat", activeSessionId: state.activeSessionId }), + makeCronJob({ + id: "rlm-2", + source: "rlm_heartbeat", + activeSessionId: state.activeSessionId, + deliveryMode: "follow_up", + }), ); expect(first).toBe("skipped"); @@ -3164,7 +3179,7 @@ describe("daemon mode helpers", () => { expect(prompt).not.toHaveBeenCalled(); }); - it("prompts idle sessions with a followUp streaming behavior to survive a mid-call stream start", async () => { + it("delivers heartbeats with a steering behavior by default", async () => { const daemon = new AgentDaemon("/tmp/prime-agent-test.sock", { defaultSessionConfig: { agentDir: "/tmp/prime-agent-test-agent", cwd: "/tmp" }, createRuntime: async () => { @@ -3172,7 +3187,12 @@ describe("daemon mode helpers", () => { }, }); const prompt = vi.fn(async () => {}); - const promptHeartbeat = vi.fn(async () => {}); + const promptHeartbeat = vi.fn( + async ( + _job: AgentCronJob, + _options?: { streamingBehavior?: "steer" | "followUp"; followUpQueueKey?: string }, + ) => {}, + ); const followUp = vi.fn(async () => true); const state = makeState("active-1") as ActiveSessionState & { runtime: ActiveSessionState["runtime"] & { @@ -3200,6 +3220,128 @@ describe("daemon mode helpers", () => { makeCronJob({ id: "heartbeat-1", source: "heartbeat", activeSessionId: state.activeSessionId }), ); + // The preparing guard adds an internal preflightResult hook. + expect(promptHeartbeat).toHaveBeenCalledWith( + expect.objectContaining({ id: "heartbeat-1", prompt: "heartbeat prompt" }), + expect.objectContaining({ + streamingBehavior: "steer", + source: "rpc", + }), + ); + expect(promptHeartbeat.mock.calls[0]?.[1]).toMatchObject({ followUpQueueKey: "heartbeat:heartbeat-1" }); + expect(prompt).not.toHaveBeenCalled(); + expect(followUp).not.toHaveBeenCalled(); + }); + + it("delivers steer heartbeats after an RPC prompt finishes preflight while its turn is still streaming", async () => { + const daemon = new AgentDaemon("/tmp/prime-agent-test.sock", { + defaultSessionConfig: { agentDir: "/tmp/prime-agent-test-agent", cwd: "/tmp" }, + createRuntime: async () => { + throw new Error("unexpected runtime creation"); + }, + }); + let releasePrompt = () => {}; + const promptFinished = new Promise((resolve) => { + releasePrompt = resolve; + }); + let reportPromptStarted = () => {}; + const promptStarted = new Promise((resolve) => { + reportPromptStarted = resolve; + }); + const sessionState = { + isStreaming: false, + isBashRunning: false, + pendingMessageCount: 0, + }; + const prompt = vi.fn(async (_message: string, options?: { preflightResult?: (didSucceed: boolean) => void }) => { + sessionState.isStreaming = true; + options?.preflightResult?.(true); + reportPromptStarted(); + await promptFinished; + }); + const promptHeartbeat = vi.fn( + async ( + _job: AgentCronJob, + options?: { streamingBehavior?: "steer" | "followUp"; preflightResult?: (didSucceed: boolean) => void }, + ) => { + options?.preflightResult?.(true); + }, + ); + const state = makeState("active-1") as ActiveSessionState & { + runtime: ActiveSessionState["runtime"] & { + session: typeof sessionState & { + prompt: typeof prompt; + promptHeartbeat: typeof promptHeartbeat; + }; + }; + }; + state.runtime.session = Object.assign(sessionState, { prompt, promptHeartbeat }) as never; + const internals = daemon as unknown as { + sessions: Map; + agentMessagePreparingTargets: Map; + promptWithAgentMessagePreparingGuard(state: ActiveSessionState, message: string): Promise; + runCronJob(job: AgentCronJob): Promise<"skipped" | undefined>; + }; + internals.sessions.set(state.activeSessionId, state); + + const promptPromise = internals.promptWithAgentMessagePreparingGuard(state, "long-running prompt"); + await promptStarted; + const preparingReleased = !internals.agentMessagePreparingTargets.has(state.activeSessionId); + const result = await internals.runCronJob( + makeCronJob({ id: "heartbeat-1", source: "heartbeat", activeSessionId: state.activeSessionId }), + ); + releasePrompt(); + await promptPromise; + + expect(preparingReleased).toBe(true); + expect(result).toBeUndefined(); + expect(promptHeartbeat).toHaveBeenCalledWith( + expect.objectContaining({ id: "heartbeat-1" }), + expect.objectContaining({ streamingBehavior: "steer" }), + ); + }); + + it("delivers follow-up heartbeats with a followUp behavior and coalescing key", async () => { + const daemon = new AgentDaemon("/tmp/prime-agent-test.sock", { + defaultSessionConfig: { agentDir: "/tmp/prime-agent-test-agent", cwd: "/tmp" }, + createRuntime: async () => { + throw new Error("unexpected runtime creation"); + }, + }); + const prompt = vi.fn(async () => {}); + const promptHeartbeat = vi.fn(async () => {}); + const followUp = vi.fn(async () => true); + const state = makeState("active-1") as ActiveSessionState & { + runtime: ActiveSessionState["runtime"] & { + session: { + isStreaming: boolean; + isBashRunning: boolean; + pendingMessageCount: number; + prompt: typeof prompt; + promptHeartbeat: typeof promptHeartbeat; + followUp: typeof followUp; + }; + }; + }; + state.runtime.session = { + isStreaming: false, + isBashRunning: false, + pendingMessageCount: 0, + prompt, + promptHeartbeat, + followUp, + } as never; + (daemon as unknown as { sessions: Map }).sessions.set(state.activeSessionId, state); + + await (daemon as unknown as { runCronJob(job: AgentCronJob): Promise<"skipped" | undefined> }).runCronJob( + makeCronJob({ + id: "heartbeat-1", + source: "heartbeat", + activeSessionId: state.activeSessionId, + deliveryMode: "follow_up", + }), + ); + // The preparing guard adds an internal preflightResult hook. expect(promptHeartbeat).toHaveBeenCalledWith( expect.objectContaining({ id: "heartbeat-1", prompt: "heartbeat prompt" }), @@ -3263,6 +3405,207 @@ describe("daemon mode helpers", () => { expect(followUp).not.toHaveBeenCalled(); }); + it("forwards queue keys when expanded daemon steer commands are queued", async () => { + const daemon = new AgentDaemon("/tmp/prime-agent-test.sock", { + defaultSessionConfig: { agentDir: "/tmp/prime-agent-test-agent", cwd: "/tmp" }, + createRuntime: async () => { + throw new Error("unexpected runtime creation"); + }, + }); + const steer = vi.fn(async () => {}); + const restoreSteeringMessage = vi.fn(async () => {}); + const state = makeState("active-1") as ActiveSessionState & { + runtime: ActiveSessionState["runtime"] & { + session: { + steer: typeof steer; + restoreSteeringMessage: typeof restoreSteeringMessage; + }; + }; + }; + state.runtime.session = { steer, restoreSteeringMessage } as never; + const internals = daemon as unknown as { + sessions: Map; + handleCommand(client: DaemonSocketClient, command: DaemonCommand): Promise; + }; + internals.sessions.set(state.activeSessionId, state); + + await internals.handleCommand(makeClient("client-1", state.activeSessionId), { + id: "command-1", + type: "steer", + activeSessionId: state.activeSessionId, + message: "queued heartbeat", + queueKey: "heartbeat:job-1", + agentMessageId: "agentmsg_steer", + }); + + expect(steer).toHaveBeenCalledWith("queued heartbeat", undefined, { + queueKey: "heartbeat:job-1", + agentMessageId: "agentmsg_steer", + }); + expect(restoreSteeringMessage).not.toHaveBeenCalled(); + }); + + it("rejects invalid heartbeat delivery modes before persisting", async () => { + const tempDir = mkdtempSync(join(tmpdir(), "prime-agent-daemon-heartbeat-delivery-mode-")); + try { + const daemon = new AgentDaemon(join(tempDir, "daemon.sock"), { + defaultSessionConfig: { agentDir: tempDir, cwd: tempDir }, + createRuntime: async () => { + throw new Error("unexpected runtime creation"); + }, + }); + const sessionFile = join(tempDir, "session.jsonl"); + const state = makeState("active-1") as ActiveSessionState & { + runtime: ActiveSessionState["runtime"] & { + session: ActiveSessionState["runtime"]["session"] & { + sessionFile: string; + sessionId: string; + }; + }; + }; + state.runtime = { + ...state.runtime, + cwd: tempDir, + session: { + sessionFile, + sessionId: "session-1", + }, + } as never; + const internals = daemon as unknown as { + cronStore: AgentCronJobStore; + sessions: Map; + handleCommand(client: DaemonSocketClient, command: DaemonCommand): Promise; + }; + internals.sessions.set(state.activeSessionId, state); + + await expect( + internals.handleCommand(makeClient("client-1", state.activeSessionId), { + id: "command-1", + type: "heartbeat_set", + activeSessionId: state.activeSessionId, + schedule: "every 5m", + prompt: "check the run", + deliveryMode: "followup" as never, + }), + ).rejects.toThrow('Heartbeat delivery mode must be "steer" or "follow_up"'); + expect(internals.cronStore.getHeartbeat(state.activeSessionId)).toBeUndefined(); + } finally { + rmSync(tempDir, { recursive: true, force: true }); + } + }); + + it("preserves the current heartbeat delivery mode when replacement omits it", async () => { + const tempDir = mkdtempSync(join(tmpdir(), "prime-agent-daemon-heartbeat-preserve-delivery-mode-")); + try { + const daemon = new AgentDaemon(join(tempDir, "daemon.sock"), { + defaultSessionConfig: { agentDir: tempDir, cwd: tempDir }, + createRuntime: async () => { + throw new Error("unexpected runtime creation"); + }, + }); + const state = makeState("active-1") as ActiveSessionState & { + runtime: ActiveSessionState["runtime"] & { + session: ActiveSessionState["runtime"]["session"] & { + removeQueuedFollowUp: ReturnType; + sessionFile: string; + sessionId: string; + }; + }; + }; + state.runtime = { + ...state.runtime, + cwd: tempDir, + session: { + removeQueuedFollowUp: vi.fn(() => false), + sessionFile: join(tempDir, "session.jsonl"), + sessionId: "session-1", + }, + } as never; + const internals = daemon as unknown as { + sessions: Map; + handleCommand( + client: DaemonSocketClient, + command: DaemonCommand, + ): Promise<{ + data: { heartbeat: AgentCronJob }; + }>; + }; + internals.sessions.set(state.activeSessionId, state); + const client = makeClient("client-1", state.activeSessionId); + + await internals.handleCommand(client, { + id: "command-1", + type: "heartbeat_set", + activeSessionId: state.activeSessionId, + schedule: "every 5m", + prompt: "first instruction", + deliveryMode: "follow_up", + }); + const replacement = await internals.handleCommand(client, { + id: "command-2", + type: "heartbeat_set", + activeSessionId: state.activeSessionId, + schedule: "every 10m", + prompt: "replacement instruction", + }); + + expect(replacement.data.heartbeat).toMatchObject({ + prompt: "replacement instruction", + deliveryMode: "follow_up", + }); + } finally { + rmSync(tempDir, { recursive: true, force: true }); + } + }); + + it("removes queued RLM heartbeat follow-ups when only delivery mode changes", () => { + const tempDir = mkdtempSync(join(tmpdir(), "prime-agent-daemon-rlm-delivery-mode-")); + try { + const daemon = new AgentDaemon(join(tempDir, "daemon.sock"), { + defaultSessionConfig: { agentDir: tempDir, cwd: tempDir }, + createRuntime: async () => { + throw new Error("unexpected runtime creation"); + }, + }); + const removeQueuedFollowUp = vi.fn(() => true); + const state = makeState("active-1") as ActiveSessionState & { + runtime: ActiveSessionState["runtime"] & { + session: ActiveSessionState["runtime"]["session"] & { + removeQueuedFollowUp: typeof removeQueuedFollowUp; + }; + }; + }; + state.runtime.session = { removeQueuedFollowUp } as never; + const internals = daemon as unknown as { + cronStore: AgentCronJobStore; + updateRlmHeartbeatForState( + state: ActiveSessionState, + input: { id: string; deliveryMode: "steer" | "follow_up" }, + ): AgentCronJob | undefined; + }; + const rlmHeartbeat = internals.cronStore.createRlmHeartbeat({ + activeSessionId: state.activeSessionId, + sessionId: "session-1", + sessionFile: join(tempDir, "session.jsonl"), + cwd: tempDir, + scheduleText: "every 5m", + prompt: "check internal state", + deliveryMode: "follow_up", + now: new Date("2026-01-01T12:00:00.000Z"), + }); + + const updated = internals.updateRlmHeartbeatForState(state, { + id: rlmHeartbeat.id, + deliveryMode: "steer", + }); + + expect(updated).toMatchObject({ id: rlmHeartbeat.id, deliveryMode: "steer" }); + expect(removeQueuedFollowUp).toHaveBeenCalledWith(`heartbeat:${rlmHeartbeat.id}`); + } finally { + rmSync(tempDir, { recursive: true, force: true }); + } + }); + it("removes queued heartbeat follow-ups when a heartbeat is cleared", async () => { const tempDir = mkdtempSync(join(tmpdir(), "prime-agent-daemon-heartbeat-clear-")); try { @@ -3561,11 +3904,17 @@ describe("daemon mode helpers", () => { }); }); -function makeCronJob(input: { id: string; source: AgentCronJob["source"]; activeSessionId: string }): AgentCronJob { +function makeCronJob(input: { + id: string; + source: AgentCronJob["source"]; + activeSessionId: string; + deliveryMode?: AgentCronJob["deliveryMode"]; +}): AgentCronJob { return { id: input.id, status: "active", source: input.source, + ...(input.deliveryMode ? { deliveryMode: input.deliveryMode } : {}), activeSessionId: input.activeSessionId, sessionId: "session-1", sessionFile: "/tmp/session.jsonl", diff --git a/packages/coding-agent/test/interactive-mode-heartbeat-command.test.ts b/packages/coding-agent/test/interactive-mode-heartbeat-command.test.ts index 48f5d0b54..a1c65e194 100644 --- a/packages/coding-agent/test/interactive-mode-heartbeat-command.test.ts +++ b/packages/coding-agent/test/interactive-mode-heartbeat-command.test.ts @@ -10,15 +10,27 @@ const interactiveModePrototype = InteractiveMode.prototype as unknown as Interac describe("InteractiveMode /heartbeat", () => { describe("argument autocomplete", () => { - it("lists the interval syntax starter for an empty prefix", () => { + it("lists the interval and delivery-mode starters for an empty prefix", () => { const items = interactiveModePrototype.getHeartbeatArgumentCompletions(""); - expect(items?.map((item) => item.label)).toEqual(["every "]); + expect(items?.map((item) => item.label)).toEqual([ + "every ", + "--steer ", + "--follow-up ", + ]); const everyItem = items?.find((item) => item.label === "every "); expect(everyItem?.value).toBe("every "); expect(everyItem?.description).toBe( "Set an interval, then add an instruction: /heartbeat every 10s Scan the logs", ); + const followUpItem = items?.find((item) => item.label === "--follow-up "); + expect(followUpItem?.value).toBe("--follow-up "); + }); + + it("filters delivery-mode flags by keyword", () => { + const items = interactiveModePrototype.getHeartbeatArgumentCompletions("--fol"); + + expect(items?.map((item) => item.label)).toEqual(["--follow-up "]); }); it("does not autocomplete lifecycle commands over free-form instructions", () => { diff --git a/packages/coding-agent/test/kernel-rlm-heartbeat-skill.test.ts b/packages/coding-agent/test/kernel-rlm-heartbeat-skill.test.ts index be05bef51..8a47a0f91 100644 --- a/packages/coding-agent/test/kernel-rlm-heartbeat-skill.test.ts +++ b/packages/coding-agent/test/kernel-rlm-heartbeat-skill.test.ts @@ -91,7 +91,7 @@ describe("RLM heartbeat skill over the kernel host bridge", () => { const manager = await provisioner.ensure(); const result = await manager.execute(` import json -created = await rlm_heartbeat.create("check tests", interval="5m", label="tests") +created = await rlm_heartbeat.create("check tests", interval="5m", label="tests", delivery_mode="follow_up") listed = await rlm_heartbeat.list(include_inactive=True) updated = await rlm_heartbeat.update(created["heartbeat"]["id"], status="pause") deleted = await rlm_heartbeat.delete(created["heartbeat"]["id"]) @@ -121,6 +121,7 @@ print(json.dumps({ instruction: "check tests", interval: "5m", label: "tests", + delivery_mode: "follow_up", }); expect(requests[1].payload).toMatchObject({ type: "rlm_heartbeat.list", include_inactive: true }); expect(requests[2].payload).toMatchObject({ type: "rlm_heartbeat.update", id: "job-1", status: "pause" }); @@ -145,4 +146,33 @@ except RuntimeError as error: 'RuntimeError: host request type "rlm_heartbeat.list" is not available in this session', ); }); + + it("rejects non-string delivery modes before calling the host", async () => { + let hostRequestCount = 0; + provisioner = new IpythonKernelProvisioner(tempDir, { + pythonSkills: [bundledRlmHeartbeatSkill()], + hostHandlers: { + "rlm_heartbeat.create": async () => { + hostRequestCount++; + return {}; + }, + }, + }); + + const manager = await provisioner.ensure(); + const result = await manager.execute(` +for value in ([], {}): + try: + await rlm_heartbeat.create("check tests", delivery_mode=value) + except TypeError as error: + print(f"TypeError: {error}") +`); + + expect(result.status).toBe("ok"); + expect(result.stdout.trim().split("\n")).toEqual([ + "TypeError: delivery_mode must be str or None, got list", + "TypeError: delivery_mode must be str or None, got dict", + ]); + expect(hostRequestCount).toBe(0); + }); }); diff --git a/packages/coding-agent/test/slash-commands.test.ts b/packages/coding-agent/test/slash-commands.test.ts index c707b6301..cc866dcc7 100644 --- a/packages/coding-agent/test/slash-commands.test.ts +++ b/packages/coding-agent/test/slash-commands.test.ts @@ -18,8 +18,9 @@ describe("built-in slash commands", () => { test("exposes heartbeat syntax guidance", () => { expect(BUILTIN_SLASH_COMMANDS.find((command) => command.name === "heartbeat")).toMatchObject({ - description: "Set or view a persistent heartbeat; supports pause, resume, stop, and clear", - argumentHint: "[status|pause|resume|stop|every ]", + description: + "Set or view a persistent heartbeat; delivery defaults to steer, use --follow-up to queue; supports pause, resume, stop, and clear", + argumentHint: "[status|pause|resume|stop|[every ] [--steer|--follow-up] ]", takesArgument: true, }); }); diff --git a/packages/coding-agent/test/suite/regressions/4257-update-restart-resume.test.ts b/packages/coding-agent/test/suite/regressions/4257-update-restart-resume.test.ts index e8ad07cc9..eb48cddee 100644 --- a/packages/coding-agent/test/suite/regressions/4257-update-restart-resume.test.ts +++ b/packages/coding-agent/test/suite/regressions/4257-update-restart-resume.test.ts @@ -22,7 +22,12 @@ type AgentDaemonUpdateInternals = { }; type QueueInternals = { - _steeringMessages: Array<{ text: string; agentMessageId?: string; message: UserMessage | CustomMessage }>; + _steeringMessages: Array<{ + text: string; + queueKey?: string; + agentMessageId?: string; + message: UserMessage | CustomMessage; + }>; _followUpMessages: Array<{ text: string; queueKey?: string; @@ -312,7 +317,9 @@ describe("issue #4257 update restart resume", () => { }; const customFollowUp = createCustomMessage("custom heartbeat"); const queueInternals = parentHarness.session as unknown as QueueInternals; - queueInternals._steeringMessages = [{ text: "queued work", agentMessageId: "agentmsg_steer", message }]; + queueInternals._steeringMessages = [ + { text: "queued work", queueKey: "heartbeat:steer", agentMessageId: "agentmsg_steer", message }, + ]; queueInternals._followUpMessages = [ { text: "heartbeat", @@ -371,7 +378,15 @@ describe("issue #4257 update restart resume", () => { clientEnv: { PRIME_SESSION: "pane-1" }, runtimeMetadata: { kind: "top-level" }, queue: { - steering: [{ message: "queued work", content, images: [image], agentMessageId: "agentmsg_steer" }], + steering: [ + { + message: "queued work", + content, + images: [image], + queueKey: "heartbeat:steer", + agentMessageId: "agentmsg_steer", + }, + ], followUp: [ { message: "heartbeat", @@ -770,6 +785,7 @@ describe("issue #4257 update restart resume", () => { activeSessionId: "active-1", message: "restored steer", content: restoredSteerContent, + queueKey: "heartbeat:steer", expandPromptTemplates: false, agentMessageId: "agentmsg_restored_steer", }), @@ -814,6 +830,7 @@ describe("issue #4257 update restart resume", () => { expect.objectContaining({ text: "restored steer", content: restoredSteerContent, + queueKey: "heartbeat:steer", agentMessageId: "agentmsg_restored_steer", }), ]); diff --git a/packages/coding-agent/test/suite/regressions/4482-heartbeat-injected-prompt.test.ts b/packages/coding-agent/test/suite/regressions/4482-heartbeat-injected-prompt.test.ts index 1aed13d7e..0c537e775 100644 --- a/packages/coding-agent/test/suite/regressions/4482-heartbeat-injected-prompt.test.ts +++ b/packages/coding-agent/test/suite/regressions/4482-heartbeat-injected-prompt.test.ts @@ -306,6 +306,55 @@ describe("ENG-4482 heartbeat injected prompt UI", () => { expect(harness.session.getFollowUpMessagePreviews()).toEqual([]); }); + it("removes queued steered heartbeat prompts by heartbeat queue key", async () => { + let releaseToolExecution: (() => void) | undefined; + const toolRelease = new Promise((resolve) => { + releaseToolExecution = resolve; + }); + const waitTool: AgentTool = { + name: "wait", + label: "Wait", + description: "Wait for release", + parameters: Type.Object({}), + execute: async () => { + await toolRelease; + return { + content: [{ type: "text", text: "released" }], + details: {}, + }; + }, + }; + const harness = await createHarness({ tools: [waitTool] }); + harnesses.push(harness); + const heartbeatText = "Check whether the long-running task needs another step."; + const heartbeatPreview = `Heartbeat prompt: ${heartbeatText}`; + harness.setResponses([ + fauxAssistantMessage(fauxToolCall("wait", {}), { stopReason: "toolUse" }), + fauxAssistantMessage("original turn complete"), + ]); + const sawToolStart = new Promise((resolve) => { + const unsubscribe = harness.session.subscribe((event) => { + if (event.type === "tool_execution_start") { + unsubscribe(); + resolve(); + } + }); + }); + + const promptPromise = harness.session.prompt("start"); + await sawToolStart; + await harness.session.promptHeartbeat(createHeartbeat(), { streamingBehavior: "steer" }); + + expect(harness.session.getSteeringMessagePreviews()).toEqual([heartbeatPreview]); + expect(harness.session.removeQueuedFollowUp("heartbeat:heartbeat-1")).toBe(true); + expect(harness.session.getSteeringMessagePreviews()).toEqual([]); + + releaseToolExecution?.(); + await promptPromise; + + expect(getUserTexts(harness)).toEqual(["start"]); + }); + it("renders heartbeat prompts as expandable injected prompt panels", () => { const component = new InjectedPromptMessageComponent(createHeartbeatPromptMessage(createHeartbeat())); const collapsed = render(component); diff --git a/packages/coding-agent/test/suite/regressions/4519-heartbeat-rebirth.test.ts b/packages/coding-agent/test/suite/regressions/4519-heartbeat-rebirth.test.ts index 6532cb3a2..a6f5e25e2 100644 --- a/packages/coding-agent/test/suite/regressions/4519-heartbeat-rebirth.test.ts +++ b/packages/coding-agent/test/suite/regressions/4519-heartbeat-rebirth.test.ts @@ -309,7 +309,7 @@ describe("ENG-4519 heartbeat rebirth", () => { expect(promptHeartbeat).toHaveBeenCalledWith( expect.objectContaining({ id: heartbeat.id, prompt: "continue scheduled work" }), expect.objectContaining({ - streamingBehavior: "followUp", + streamingBehavior: "steer", followUpQueueKey: `heartbeat:${heartbeat.id}`, source: "rpc", }), -- 2.54.0