diff --git a/README.md b/README.md index 5717c52..b12de2a 100644 --- a/README.md +++ b/README.md @@ -252,6 +252,7 @@ Then invite the app to the channel and pair in that channel/thread with `relay p | select session | `/use ` | `relay use ` | `/relay use ` | | forget offline session | `/forget ` | `relay forget ` | `/relay forget ` | | one-shot prompt | `/to ` | `relay to ` | `/relay to ` | +| start replacement session | `/new` | `relay new` | `/relay new` | | progress mode | `/progress ` | `relay progress ` | `/relay progress ` | | current progress mode | `/progress` | `relay progress` | `/relay progress` | | alias current session | `/alias ` | `relay alias ` | `/relay alias ` | @@ -278,6 +279,8 @@ Then invite the app to the channel and pair in that channel/thread with `relay p Remote `/disconnect` is scoped to the requesting chat/conversation only: it revokes that Telegram, Discord, or Slack binding and suppresses future session output/buttons there, without disconnecting other messengers that remain paired to the same Pi session. Local `/relay disconnect` is broader and disconnects the current session from all paired messenger bindings. +Remote `/new` starts a replacement session only for the selected online, idle Pi route; it does not start or resume an offline Pi process. Busy routes refuse the request and suggest waiting or using `/abort`. When Pi provides safe command-capable session control, eligible requester bindings and active selections move to the replacement session automatically; unsupported routes return an explicit limitation. Local `/new` uses the same short, strict same-machine/same-workspace handoff and otherwise falls back to normal offline/reconnect behavior. + Remote `send-file` is requester-scoped: an authorized Telegram/Discord/Slack user may request a workspace-relative, validated path and PiRelay uploads it only back to that same conversation/thread. Targeted fan-out remains local-only via `/relay send-file [caption]`; remote forms must not include messenger targets such as `all` or `slack`. Remote skill invocation is opt-in via `skills.enabled`. When enabled, `/skills` lists allowed local Pi skills from the live session command registry, and `/skill [input]` asks Pi to use that skill. If input is omitted, the next message from the same authorized requester is used as skill input; `/cancel` or `/skill cancel` cancels the pending input. diff --git a/docs/adapters.md b/docs/adapters.md index 0c5ae61..60e17bd 100644 --- a/docs/adapters.md +++ b/docs/adapters.md @@ -45,7 +45,7 @@ Inbound messenger events are normalized as messages or actions before relay hand `/relay` is canonical for local Pi commands. The old `/telegram-tunnel` local namespace has been removed. -Remote messenger adapters expose the same PiRelay command semantics wherever the platform allows text commands, slash commands, buttons, or fallbacks: `/help`, `/status`, `/sessions`, `/use`, `/to`, `/alias`, `/forget`, `/progress`, `/recent`, `/summary`, `/full`, `/skills`, `/skill`, `/cancel`, `/images`, `/send-image`, `/send-file`, `/steer`, `/followup`, `/abort`, `/compact`, `/pause`, `/resume`, and `/disconnect`. Local fan-out file delivery remains local-only via `/relay send-file [caption]`; remote `send-file`/`sendfile` requests are requester-scoped, validated against the workspace, and delivered only back to the same authorized conversation/thread. If an adapter cannot perform a command because of a declared capability limit, it should return an explicit limitation instead of falling through to generic unsupported-command help. +Remote messenger adapters expose the same PiRelay command semantics wherever the platform allows text commands, slash commands, buttons, or fallbacks: `/help`, `/status`, `/sessions`, `/use`, `/to`, `/new`, `/alias`, `/forget`, `/progress`, `/recent`, `/summary`, `/full`, `/skills`, `/skill`, `/cancel`, `/images`, `/send-image`, `/send-file`, `/steer`, `/followup`, `/abort`, `/compact`, `/pause`, `/resume`, and `/disconnect`. Remote `/new` targets only the selected online idle route; it never starts an offline Pi process, refuses busy routes, and reports an explicit capability limitation when command-capable session control is unavailable. Local fan-out file delivery remains local-only via `/relay send-file [caption]`; remote `send-file`/`sendfile` requests are requester-scoped, validated against the workspace, and delivered only back to the same authorized conversation/thread. If an adapter cannot perform a command because of a declared capability limit, it should return an explicit limitation instead of falling through to generic unsupported-command help. Discord is special because Discord owns the `/...` application-command UI. The reliable Discord baseline is ordinary DM text with a prefix, for example `relay status`, `relay sessions`, `relay full`, `relay abort`, and `relay pair 123-456` during pairing. PiRelay best-effort syncs one namespaced native `/relay` command with subcommands when Discord application commands are available, but setup should still advertise `relay ` first; bare `/status`-style Discord aliases remain best-effort conveniences only when Discord delivers them as message text, and `/start ` is accepted for pairing compatibility. diff --git a/docs/testing.md b/docs/testing.md index 241116d..7561453 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -87,6 +87,8 @@ Check: 8. Use `/use docs`, then set `/alias phone`, and verify `/sessions` and selectors use the alias while retaining the original label for disambiguation. 9. Use `/progress quiet`, `/progress verbose`, and `/recent` during a run and verify safe progress noise changes without suppressing terminal completion/failure/abort notifications. 10. Reconnect the disconnected session and verify local prompts still work there too. +11. Run local `/new` from a paired idle session and verify the old route unregisters immediately, no false offline notification is sent, the same-workspace replacement receives the binding/active selection, old buttons become stale, and the messenger receives a moved notification. +12. From authorized Telegram, send `/new` to an online idle selected route and verify relay control moves to the replacement session; repeat while busy, offline, ambiguous, paused, and unsupported, and verify no unsafe session replacement or binding mutation occurs. ## 5. Image bridge behavior @@ -122,7 +124,7 @@ These adapter foundations are DM-first and use channel-specific credentials/conf 9. Run `/relay connect discord docs`, scan the QR bot profile/DM link when `discord.applicationId` is configured, then DM the bot `relay pair ` before expiry (`/start ` remains a compatibility alias). Confirm the pairing locally in Pi, optionally choosing to trust the Discord user. If the bot cannot be DM'd, check shared-server membership, server member DM privacy settings, and that the bot was already invited with the `bot` scope during setup. 10. After pairing, send `relay status` and `relay sessions` in the Discord DM and verify the same core fields/semantics as Telegram: safe session label, online state, busy state, model, progress mode, last activity, active marker, and no raw session file path or binding storage key. If Discord command sync has completed, also try `/relay status`; bare `/status` and `/sessions` may work when Discord delivers them as text, but smoke tests should not depend on those aliases. 11. Send a normal Discord DM prompt while Pi is idle and verify it reaches the current Pi session and the final Pi completion returns to Discord; repeat while busy and verify the configured busy delivery acknowledgement plus terminal completion/failure/abort notification. -12. Exercise Discord command parity using the reliable prefix forms: `relay use`, `relay to`, `relay alias`, `relay progress`, `relay recent`, `relay summary`, `relay full`, `relay skills`, `relay skill`, `relay cancel`, `relay images`, `relay send-image`, `relay send-file`, `relay steer`, `relay followup`, `relay abort`, `relay compact`, `relay pause`, `relay resume`, and `relay disconnect`; verify commands either work with Telegram-equivalent semantics or return an explicit capability/configuration limitation, not generic unsupported-command help. +12. Exercise Discord command parity using the reliable prefix forms: `relay use`, `relay to`, `relay new`, `relay alias`, `relay progress`, `relay recent`, `relay summary`, `relay full`, `relay skills`, `relay skill`, `relay cancel`, `relay images`, `relay send-image`, `relay send-file`, `relay steer`, `relay followup`, `relay abort`, `relay compact`, `relay pause`, `relay resume`, and `relay disconnect`; verify commands either work with Telegram-equivalent semantics or return an explicit capability/configuration limitation, not generic unsupported-command help. 13. Enable Discord guild-channel control without `allowGuildIds`; verify `/relay doctor` reports an actionable warning/error and `/relay connect discord` refuses pairing until fixed. 14. Run `/relay setup slack` for `eventMode: "socket"` and verify it recommends Socket Mode, requires an app-level token with `connections:write`, exposes a copyable App manifest tab/action, includes interactivity and `/relay`, explains App Home Messages Tab plus `message.im`/`im:history`/`im:read` for DMs, `reactions:write` for thinking indicators, and `files:write` for image/file delivery, and warns when workspace, App ID, or bot-user identity cannot be established; switch to `eventMode: "webhook"` without a signing secret and verify doctor reports the webhook signing requirement. 15. Run `/relay connect slack docs` with enabled mock config and verify the displayed pairing instruction is time-limited, channel-specific, short PIN-style, highlighted in the QR dialog, and copyable with `c`; when `slack.appId` is configured or discovered from `auth.test`, verify the TUI renders a Slack App Home QR/open link and clearly offers both paths: open App Home DM via QR/link or paste the command directly in an invited Slack channel/thread after enabling `slack.allowChannelMessages`. diff --git a/extensions/relay/adapters/discord/runtime.ts b/extensions/relay/adapters/discord/runtime.ts index 457c9d3..615a7a8 100644 --- a/extensions/relay/adapters/discord/runtime.ts +++ b/extensions/relay/adapters/discord/runtime.ts @@ -13,7 +13,7 @@ import { displayProgressMode, formatProgressUpdate, normalizeProgressMode, progr import { deliverLiveProgress, type LiveProgressDeliveryState } from "../../notifications/progress-delivery.js"; import { sendFinalOutputWithFallback } from "../../core/final-output.js"; import { formatRelayLifecycleNotification, type RelayLifecycleEventKind } from "../../notifications/lifecycle.js"; -import { abortRouteSafely, compactRouteSafely, deliverRoutePrompt, latestRouteImagesSafely, probeRouteAvailability, routeActionDisplayMessage, routeIdleState, routeImageByPathSafely, routeModelState, routeSkillCommandsSafely, routeWorkspaceRootSafely, unavailableRouteMessage } from "../../core/route-actions.js"; +import { abortRouteSafely, compactRouteSafely, deliverRoutePrompt, latestRouteImagesSafely, newSessionRouteActionMessage, newSessionRouteSafely, probeRouteAvailability, routeActionDisplayMessage, routeIdleState, routeImageByPathSafely, routeModelState, routeSkillCommandsSafely, routeWorkspaceRootSafely, unavailableRouteMessage } from "../../core/route-actions.js"; import { statusSnapshotForRoute } from "../../core/relay-core.js"; import { authorityOutcomeAllowsDelivery, bindingAuthorityDiagnostic, channelDestinationKey, resolveChannelBindingAuthority } from "../../core/binding-authority.js"; import { redactSecrets } from "../../config/setup.js"; @@ -902,6 +902,15 @@ export class DiscordRuntime { case "help": await this.sendText(message, DISCORD_HELP_TEXT); return; + case "new": { + const outcome = await newSessionRouteSafely(route, this.discordRequester(route, message)); + if (outcome.kind !== "success") { + await this.sendText(message, newSessionRouteActionMessage(outcome)); + return; + } + await this.sendText(message, "New Pi session started. Relay control is moving to the replacement session."); + return; + } case "status": await this.sendText(message, this.statusTextForRoute(route, binding, true)); return; @@ -1587,7 +1596,7 @@ export class DiscordRuntime { } } -export type DiscordCommandName = "help" | "status" | "sessions" | "use" | "to" | "progress" | "notify" | "alias" | "forget" | "recent" | "activity" | "summary" | "full" | "skills" | "skill" | "cancel" | "images" | "send-file" | "sendfile" | "send-image" | "sendimage" | "steer" | "followup" | "abort" | "compact" | "pause" | "resume" | "disconnect"; +export type DiscordCommandName = "help" | "status" | "sessions" | "use" | "to" | "new" | "progress" | "notify" | "alias" | "forget" | "recent" | "activity" | "summary" | "full" | "skills" | "skill" | "cancel" | "images" | "send-file" | "sendfile" | "send-image" | "sendimage" | "steer" | "followup" | "abort" | "compact" | "pause" | "resume" | "disconnect"; type DiscordCommand = { name: DiscordCommandName; args: string }; @@ -1597,6 +1606,7 @@ export const DISCORD_SUPPORTED_COMMANDS: readonly DiscordCommandName[] = [ "sessions", "use", "to", + "new", "progress", "notify", "alias", diff --git a/extensions/relay/adapters/slack/runtime.ts b/extensions/relay/adapters/slack/runtime.ts index 2b68ad2..ad320aa 100644 --- a/extensions/relay/adapters/slack/runtime.ts +++ b/extensions/relay/adapters/slack/runtime.ts @@ -13,7 +13,7 @@ import { displayProgressMode, formatProgressUpdate, normalizeProgressMode, progr import { deliverLiveProgress, type LiveProgressDeliveryState } from "../../notifications/progress-delivery.js"; import { sendFinalOutputWithFallback } from "../../core/final-output.js"; import { deliverWorkspaceFileToRequester, formatRequesterFileDeliveryResult, parseRemoteSendFileArgs, type RelayFileDeliveryRequester } from "../../core/requester-file-delivery.js"; -import { abortRouteSafely, compactRouteSafely, deliverRoutePrompt, latestRouteImagesSafely, probeRouteAvailability, routeActionDisplayMessage, routeIdleState, routeImageByPathSafely, routeModelState, routeSkillCommandsSafely, routeWorkspaceRootSafely, unavailableRouteMessage } from "../../core/route-actions.js"; +import { abortRouteSafely, compactRouteSafely, deliverRoutePrompt, latestRouteImagesSafely, newSessionRouteActionMessage, newSessionRouteSafely, probeRouteAvailability, routeActionDisplayMessage, routeIdleState, routeImageByPathSafely, routeModelState, routeSkillCommandsSafely, routeWorkspaceRootSafely, unavailableRouteMessage } from "../../core/route-actions.js"; import { statusSnapshotForRoute } from "../../core/relay-core.js"; import { authorityOutcomeAllowsDelivery, bindingAuthorityDiagnostic, channelDestinationKey, resolveChannelBindingAuthority } from "../../core/binding-authority.js"; import { formatRelayLifecycleNotification, type RelayLifecycleEventKind } from "../../notifications/lifecycle.js"; @@ -956,6 +956,15 @@ export class SlackRuntime { case "help": await this.sendText(message, SLACK_HELP_TEXT); return; + case "new": { + const outcome = await newSessionRouteSafely(route, this.slackRequester(route, message)); + if (outcome.kind !== "success") { + await this.sendText(message, newSessionRouteActionMessage(outcome)); + return; + } + await this.sendText(message, "New Pi session started. Relay control is moving to the replacement session."); + return; + } case "status": { const availability = slackRouteAvailability(route); await this.sendText(message, formatRelayStatusForRoute(route, { online: availability.online, busy: availability.busy, binding })); diff --git a/extensions/relay/adapters/telegram/runtime.ts b/extensions/relay/adapters/telegram/runtime.ts index 81dbd54..97a1a08 100644 --- a/extensions/relay/adapters/telegram/runtime.ts +++ b/extensions/relay/adapters/telegram/runtime.ts @@ -1,7 +1,7 @@ import { writeFile } from "node:fs/promises"; import lockfile from "proper-lockfile"; import { ensureParentDir, ensureStateDir, getLockFilePath } from "../../state/paths.js"; -import { abortRouteSafely, compactRouteSafely, deliverRoutePrompt, latestRouteImagesSafely, probeRouteAvailability, routeActionDisplayMessage, routeIdleState, routeImageByPathSafely, routeModelState, routeSkillCommandsSafely, routeWorkspaceRootSafely, unavailableRouteMessage } from "../../core/route-actions.js"; +import { abortRouteSafely, compactRouteSafely, deliverRoutePrompt, latestRouteImagesSafely, newSessionRouteActionMessage, newSessionRouteSafely, probeRouteAvailability, routeActionDisplayMessage, routeIdleState, routeImageByPathSafely, routeModelState, routeSkillCommandsSafely, routeWorkspaceRootSafely, unavailableRouteMessage } from "../../core/route-actions.js"; import { statusSnapshotForRoute } from "../../core/relay-core.js"; import { authorityOutcomeAllowsDelivery, bindingAuthorityDiagnostic, resolveTelegramBindingAuthority, telegramDestinationKey } from "../../core/binding-authority.js"; import { BrokerTunnelRuntime } from "../../broker/tunnel-runtime.js"; @@ -1849,6 +1849,15 @@ export class InProcessTunnelRuntime implements TunnelRuntime { } switch (command) { + case "new": { + const outcome = await newSessionRouteSafely(route, this.telegramRequester(route, message)); + if (outcome.kind !== "success") { + await this.api.sendPlainText(message.chat.id, newSessionRouteActionMessage(outcome)); + return; + } + await this.api.sendPlainText(message.chat.id, "New Pi session started. Relay control is moving to the replacement session."); + return; + } case "status": { await this.sendTextWithKeyboard(message.chat.id, this.statusTextForRoute(route, true), this.dashboardKeyboardForRoute(route)); return; diff --git a/extensions/relay/broker/process.js b/extensions/relay/broker/process.js index a890f6e..92d059a 100644 --- a/extensions/relay/broker/process.js +++ b/extensions/relay/broker/process.js @@ -1908,6 +1908,26 @@ async function handleAuthorizedCommand(message, route, command, args) { } switch (command) { + case 'new': { + if (isEffectivelyBusy(route)) { + await sendPlainText(message.chat.id, 'Pi is busy. Use /abort or wait for the active work to finish before starting a new session.'); + return; + } + try { + const outcome = await requestClient(route, 'newSession', { requester: requesterForMessage(route, message) }); + if (outcome?.kind === 'success') { + await sendPlainText(message.chat.id, 'New Pi session started. Relay control is moving to the replacement session.'); + return; + } + const text = outcome?.kind === 'failed' + ? typeof outcome.safeMessage === 'string' ? outcome.safeMessage : 'Could not start a new Pi session.' + : typeof outcome?.message === 'string' ? outcome.message : 'The Pi session is unavailable. Resume it locally, then try again.'; + await sendPlainText(message.chat.id, text); + } catch { + await sendPlainText(message.chat.id, 'The Pi session became unavailable before a new session could be started. Use /sessions to check its status.'); + } + return; + } case 'status': { await sendPlainText(message.chat.id, statusTextForRoute(route), dashboardKeyboardForRoute(route)); return; diff --git a/extensions/relay/broker/tunnel-runtime.ts b/extensions/relay/broker/tunnel-runtime.ts index 589b3fb..18a01d4 100644 --- a/extensions/relay/broker/tunnel-runtime.ts +++ b/extensions/relay/broker/tunnel-runtime.ts @@ -10,7 +10,7 @@ import { deliverWorkspaceFileToRequester, formatRequesterFileDeliveryResult, typ import { ensureStateDir } from "../state/paths.js"; import { TunnelStateStore } from "../state/tunnel-store.js"; import { relayRouteStateForRoute, statusSnapshotForRoute, type RelayRouteState } from "../core/relay-core.js"; -import { abortRouteSafely, compactRouteSafely, deliverRoutePrompt, latestRouteImagesSafely, routeActionDisplayMessage, routeImageByPathSafely, routeSkillCommandsSafely, routeWorkspaceRootSafely, unavailableRouteMessage } from "../core/route-actions.js"; +import { abortRouteSafely, compactRouteSafely, deliverRoutePrompt, latestRouteImagesSafely, newSessionRouteSafely, routeActionDisplayMessage, routeImageByPathSafely, routeSkillCommandsSafely, routeWorkspaceRootSafely, unavailableRouteMessage } from "../core/route-actions.js"; import { relayPipelineProtocolVersion } from "../middleware/pipeline.js"; import { sha256 } from "../core/utils.js"; import { brokerScopeControlPaths, ensureScopedBroker, normalizeBrokerNamespace } from "./supervisor.js"; @@ -382,6 +382,12 @@ export class BrokerTunnelRuntime implements TunnelRuntime { await respond({ ok: true, result: outcome.result }); return; } + case "newSession": { + const requester = isRecord(request.requester) ? request.requester as unknown as RelayFileDeliveryRequester : undefined; + const outcome = await newSessionRouteSafely(route, requester); + await respond({ ok: true, result: outcome.kind === "failed" ? { kind: "failed", safeMessage: outcome.safeMessage } : outcome }); + return; + } case "abort": { const outcome = abortRouteSafely(route); if (outcome.kind !== "success") { diff --git a/extensions/relay/commands/remote.ts b/extensions/relay/commands/remote.ts index d87abe5..58490fc 100644 --- a/extensions/relay/commands/remote.ts +++ b/extensions/relay/commands/remote.ts @@ -19,6 +19,7 @@ export const CANONICAL_REMOTE_COMMANDS = [ { command: "use", usage: "/use ", description: "select an active session", allowedWhilePaused: true }, { command: "forget", usage: "/forget ", description: "remove an offline session from the list", allowedWhilePaused: true }, { command: "to", usage: "/to ", description: "send one prompt without switching sessions" }, + { command: "new", usage: "/new", description: "start a replacement session for the selected live Pi route" }, { command: "summary", usage: "/summary", description: "latest summary/excerpt" }, { command: "full", usage: "/full", description: "latest full assistant output" }, { command: "skills", usage: "/skills", description: "list remote-invokable local skills", allowedWhilePaused: true }, diff --git a/extensions/relay/core/binding-authority.ts b/extensions/relay/core/binding-authority.ts index 06c361d..57f974c 100644 --- a/extensions/relay/core/binding-authority.ts +++ b/extensions/relay/core/binding-authority.ts @@ -107,6 +107,7 @@ export function bindingAuthorityDiagnostic(outcome: BindingAuthorityOutcome { + if (binding.movedToSessionKey) return { kind: "moved", binding }; if (binding.status === "revoked" || binding.revokedAt) return { kind: "revoked", binding }; if (!telegramMatches(binding, expected)) return { kind: "moved", binding }; if (binding.paused && !expected.includePaused) return { kind: "paused", binding }; @@ -114,6 +115,7 @@ function classifyTelegramPersisted(binding: PersistedBindingRecord, expected: Te } function classifyChannelPersisted(binding: ChannelPersistedBindingRecord, expected: ChannelBindingAuthorityExpected): BindingAuthorityOutcome { + if (binding.movedToSessionKey) return { kind: "moved", binding }; const instanceId = expected.instanceId ?? "default"; if (binding.status === "revoked" || binding.revokedAt) return { kind: "revoked", binding }; if (binding.channel !== expected.channel || (binding.instanceId ?? "default") !== instanceId || binding.sessionKey !== expected.sessionKey) return { kind: "moved", binding }; diff --git a/extensions/relay/core/index.ts b/extensions/relay/core/index.ts index 5394052..4d772b8 100644 --- a/extensions/relay/core/index.ts +++ b/extensions/relay/core/index.ts @@ -5,3 +5,4 @@ export * from "./agent-delegation.js"; export * from "./agent-delegation-approval.js"; export type * from "./adapter-contracts.js"; export type * from "./session-contracts.js"; +export * from "./session-handoff.js"; diff --git a/extensions/relay/core/route-actions.ts b/extensions/relay/core/route-actions.ts index 477cef8..bb3aecb 100644 --- a/extensions/relay/core/route-actions.ts +++ b/extensions/relay/core/route-actions.ts @@ -149,6 +149,34 @@ export async function deliverRoutePrompt(route: SessionRoute, options: RouteProm return routeActionSuccess({ idle: probe.idle, deliverAs }); } +export type NewSessionRouteActionOutcome = + | { kind: "success"; result: { started: true } } + | { kind: "unavailable"; message: string } + | { kind: "busy"; message: string } + | { kind: "unsupported"; message: string } + | { kind: "cancelled"; message: string } + | { kind: "failed"; error: unknown; safeMessage: string }; + +export async function newSessionRouteSafely(route: SessionRoute, requester?: RelayFileDeliveryRequester): Promise { + const probe = probeRouteAvailability(route); + if (probe.kind === "unavailable") return probe; + if (probe.busy) return { kind: "busy", message: "Pi is busy. Use /abort or wait for the active work to finish before starting a new session." }; + if (!route.actions.newSession) return { kind: "unsupported", message: "This Pi route cannot start a new session remotely. Run /new locally instead." }; + + try { + const result = await route.actions.newSession(requester); + if (result.cancelled) return { kind: "cancelled", message: "The new-session request was cancelled locally." }; + return { kind: "success", result: { started: true } }; + } catch (error) { + const outcome = routeActionOutcomeFromError(error, "Could not start a new Pi session."); + return outcome.kind === "unavailable" ? outcome : { kind: "failed", error: outcome.error, safeMessage: outcome.safeMessage }; + } +} + +export function newSessionRouteActionMessage(outcome: Exclude): string { + return outcome.kind === "failed" ? outcome.safeMessage : outcome.message; +} + export type RouteControlOperationOutcome = RouteActionOutcome; export function abortRouteSafely(route: SessionRoute, options: { alreadyIdleMessage?: string; safeFailureMessage?: string } = {}): RouteControlOperationOutcome { diff --git a/extensions/relay/core/session-handoff.ts b/extensions/relay/core/session-handoff.ts new file mode 100644 index 0000000..a63b898 --- /dev/null +++ b/extensions/relay/core/session-handoff.ts @@ -0,0 +1,170 @@ +import { randomUUID } from "node:crypto"; +import { resolve } from "node:path"; +import type { ChannelBinding } from "./channel-adapter.js"; +import type { ChannelActiveSelectionRecord, ChannelPersistedBindingRecord, PersistedBindingRecord, TunnelStoreData } from "./types.js"; + +export type SessionHandoffReason = "local-new" | "remote-new"; + +export interface SessionHandoffBindingSummary { + channel: ChannelBinding["channel"]; + instanceId: string; + conversationId: string; + userId: string; + paused: boolean; +} + +export interface PendingSessionHandoff { + id: string; + oldSessionKey: string; + oldSessionId: string; + oldSessionFile?: string; + oldSessionLabel: string; + runtimeInstanceId: string; + machineId: string; + workspaceRoot: string; + reason: SessionHandoffReason; + requester?: SessionHandoffBindingSummary; + bindings: SessionHandoffBindingSummary[]; + activeSelections: ChannelActiveSelectionRecord[]; + createdAt: number; + expiresAt: number; + explicitDisconnect?: boolean; +} + +export interface ReplacementSessionIdentity { + sessionKey: string; + runtimeInstanceId: string; + machineId: string; + workspaceRoot: string; +} + +export type PendingHandoffMatch = + | { kind: "matched"; handoff: PendingSessionHandoff } + | { kind: "none" } + | { kind: "ambiguous"; candidates: PendingSessionHandoff[] }; + +interface RegisteredPendingHandoff { + handoff: PendingSessionHandoff; + timeout?: ReturnType; + onExpire?: (handoff: PendingSessionHandoff) => void | Promise; +} + +const pendingHandoffs = new Map(); +export const relayRuntimeInstanceId = randomUUID(); +export const DEFAULT_SESSION_HANDOFF_TTL_MS = 5_000; + +export function normalizeWorkspaceRoot(path: string): string { + return resolve(path); +} + +export function createPendingSessionHandoff(input: Omit & { + workspaceRoot: string; + now?: number; + ttlMs?: number; +}): PendingSessionHandoff { + const createdAt = input.now ?? Date.now(); + return { + ...input, + id: randomUUID(), + workspaceRoot: normalizeWorkspaceRoot(input.workspaceRoot), + createdAt, + expiresAt: createdAt + (input.ttlMs ?? DEFAULT_SESSION_HANDOFF_TTL_MS), + }; +} + +export function bindingSummariesForSession(data: TunnelStoreData, sessionKey: string): SessionHandoffBindingSummary[] { + const summaries: SessionHandoffBindingSummary[] = []; + const telegram = data.bindings[sessionKey]; + if (telegram?.status === "active") summaries.push(telegramBindingSummary(telegram)); + for (const binding of Object.values(data.channelBindings)) { + if (binding.sessionKey === sessionKey && binding.status === "active") summaries.push(channelBindingSummary(binding)); + } + return summaries; +} + +export function activeSelectionsForSession(data: TunnelStoreData, sessionKey: string): ChannelActiveSelectionRecord[] { + return Object.values(data.activeChannelSelections).filter((selection) => selection.sessionKey === sessionKey); +} + +export function matchPendingSessionHandoffs(candidates: readonly PendingSessionHandoff[], replacement: ReplacementSessionIdentity, now = Date.now()): PendingHandoffMatch { + const workspaceRoot = normalizeWorkspaceRoot(replacement.workspaceRoot); + const matches = candidates.filter((candidate) => !candidate.explicitDisconnect + && candidate.expiresAt > now + && candidate.oldSessionKey !== replacement.sessionKey + && candidate.runtimeInstanceId === replacement.runtimeInstanceId + && candidate.machineId === replacement.machineId + && candidate.workspaceRoot === workspaceRoot + && candidate.bindings.length > 0); + if (matches.length === 0) return { kind: "none" }; + if (matches.length > 1) return { kind: "ambiguous", candidates: matches }; + return { kind: "matched", handoff: matches[0]! }; +} + +export function registerPendingSessionHandoff(handoff: PendingSessionHandoff, onExpire?: (handoff: PendingSessionHandoff) => void | Promise): void { + removePendingSessionHandoff(handoff.id); + const delay = Math.max(0, handoff.expiresAt - Date.now()); + const timeout = onExpire ? setTimeout(async () => { + await expirePendingSessionHandoff(handoff.id); + }, delay) : undefined; + timeout?.unref?.(); + pendingHandoffs.set(handoff.id, { handoff, timeout, onExpire }); +} + +export function findPendingSessionHandoff(replacement: ReplacementSessionIdentity, now = Date.now()): PendingHandoffMatch { + pruneExpiredPendingSessionHandoffs(now); + return matchPendingSessionHandoffs([...pendingHandoffs.values()].map((entry) => entry.handoff), replacement, now); +} + +export function takePendingSessionHandoff(replacement: ReplacementSessionIdentity, now = Date.now()): PendingHandoffMatch { + const match = findPendingSessionHandoff(replacement, now); + if (match.kind === "matched") removePendingSessionHandoff(match.handoff.id); + return match; +} + +export function removePendingSessionHandoff(id: string): PendingSessionHandoff | undefined { + const entry = pendingHandoffs.get(id); + if (!entry) return undefined; + if (entry.timeout) clearTimeout(entry.timeout); + pendingHandoffs.delete(id); + return entry.handoff; +} + +export function removePendingSessionHandoffsForSession(sessionKey: string): PendingSessionHandoff[] { + const removed: PendingSessionHandoff[] = []; + for (const entry of [...pendingHandoffs.values()]) { + if (entry.handoff.oldSessionKey !== sessionKey) continue; + const handoff = removePendingSessionHandoff(entry.handoff.id); + if (handoff) removed.push(handoff); + } + return removed; +} + +export function listPendingSessionHandoffs(): PendingSessionHandoff[] { + pruneExpiredPendingSessionHandoffs(); + return [...pendingHandoffs.values()].map((entry) => entry.handoff); +} + +function pruneExpiredPendingSessionHandoffs(now = Date.now()): void { + void expirePendingSessionHandoffs(now); +} + +export async function expirePendingSessionHandoffs(now = Date.now()): Promise { + const expired = [...pendingHandoffs.values()].filter((entry) => entry.handoff.expiresAt <= now); + await Promise.all(expired.map((entry) => expirePendingSessionHandoff(entry.handoff.id))); +} + +async function expirePendingSessionHandoff(id: string): Promise { + const entry = pendingHandoffs.get(id); + if (!entry) return; + if (entry.timeout) clearTimeout(entry.timeout); + pendingHandoffs.delete(id); + if (entry.onExpire) await Promise.resolve(entry.onExpire(entry.handoff)).catch(() => undefined); +} + +function telegramBindingSummary(binding: PersistedBindingRecord): SessionHandoffBindingSummary { + return { channel: "telegram", instanceId: "default", conversationId: String(binding.chatId), userId: String(binding.userId), paused: Boolean(binding.paused) }; +} + +function channelBindingSummary(binding: ChannelPersistedBindingRecord): SessionHandoffBindingSummary { + return { channel: binding.channel, instanceId: binding.instanceId ?? "default", conversationId: binding.conversationId, userId: binding.userId, paused: Boolean(binding.paused) }; +} diff --git a/extensions/relay/core/types.ts b/extensions/relay/core/types.ts index 05761ba..50a6079 100644 --- a/extensions/relay/core/types.ts +++ b/extensions/relay/core/types.ts @@ -132,7 +132,12 @@ export interface TelegramChatSummary { title?: string; } -export interface TelegramBindingMetadata { +export interface MovedSessionBindingMetadata { + movedToSessionKey?: string; + movedAt?: string; +} + +export interface TelegramBindingMetadata extends MovedSessionBindingMetadata { sessionKey: string; sessionId: string; sessionFile?: string; @@ -181,7 +186,7 @@ export interface SetupCache { validatedAt: string; } -export interface ChannelPersistedBindingRecord extends ChannelBinding { +export interface ChannelPersistedBindingRecord extends ChannelBinding, MovedSessionBindingMetadata { status: "active" | "revoked"; } @@ -289,6 +294,8 @@ export interface LatestTurnImageMetadata { fileCount?: number; } +export type NewSessionActionResult = { cancelled: boolean }; + export interface SessionRouteActions { /** @deprecated Prefer narrow lifetime-safe helpers on this object. */ context: ExtensionContext; @@ -307,6 +314,8 @@ export interface SessionRouteActions { refreshLocalStatus?(): void; persistBinding(binding: TelegramBindingMetadata | null, revoked?: boolean): void; promptLocalConfirmation(identity: RelayPairingIdentity): Promise; + newSession?(requester?: RelayFileDeliveryRequester): Promise; + abort(): void; compact(): Promise; resolveApprovalDecision?(decision: ApprovalDecisionRequest): Promise; diff --git a/extensions/relay/notifications/lifecycle.ts b/extensions/relay/notifications/lifecycle.ts index 24229e1..583dddc 100644 --- a/extensions/relay/notifications/lifecycle.ts +++ b/extensions/relay/notifications/lifecycle.ts @@ -1,6 +1,6 @@ import type { ChannelAdapterKind } from "../core/channel-adapter.js"; -export type RelayLifecycleEventKind = "offline" | "online" | "disconnected"; +export type RelayLifecycleEventKind = "offline" | "online" | "disconnected" | "moved"; export type RelayLifecycleState = "online" | "offline" | "disconnected"; export interface RelayLifecycleNotificationRecord { @@ -30,6 +30,8 @@ export function formatRelayLifecycleNotification(input: { kind: RelayLifecycleEv return `${subject} went offline locally. Restart Pi to resume relay control.`; case "online": return `${subject} is back online.`; + case "moved": + return `${subject} replaced the previous session. Relay control moved here automatically.`; case "disconnected": return input.channel === "slack" ? `PiRelay was disconnected locally for ${subject}. This chat is no longer paired; run \`relay pair \` from a fresh local pairing to reconnect.` @@ -78,12 +80,13 @@ export function decideRelayLifecycleNotification(input: { } function lifecycleStateForEvent(kind: RelayLifecycleEventKind): RelayLifecycleState { - if (kind === "online") return "online"; + if (kind === "online" || kind === "moved") return "online"; if (kind === "offline") return "offline"; return "disconnected"; } function shouldNotifyLifecycle(input: { previous?: RelayLifecycleNotificationRecord; kind: RelayLifecycleEventKind; nowIso: string; debounceMs: number }): boolean { + if (input.kind === "moved") return !input.previous || !withinDebounce(input.previous, input.kind, input.nowIso, input.debounceMs); if (input.kind === "online") return input.previous?.state === "offline" && !withinDebounce(input.previous, input.kind, input.nowIso, input.debounceMs); if (!input.previous) return true; if (input.previous.state !== lifecycleStateForEvent(input.kind)) return true; diff --git a/extensions/relay/runtime/extension-runtime.ts b/extensions/relay/runtime/extension-runtime.ts index 647980c..82a7447 100644 --- a/extensions/relay/runtime/extension-runtime.ts +++ b/extensions/relay/runtime/extension-runtime.ts @@ -1,6 +1,6 @@ import type { AgentMessage } from "@mariozechner/pi-agent-core"; import { Type, type ImageContent } from "@mariozechner/pi-ai"; -import type { ExtensionAPI, ExtensionContext, Theme } from "@mariozechner/pi-coding-agent"; +import type { ExtensionAPI, ExtensionCommandContext, ExtensionContext, Theme } from "@mariozechner/pi-coding-agent"; import { matchesKey, visibleWidth } from "@mariozechner/pi-tui"; import { loadTelegramTunnelConfig, ConfigError } from "../config/tunnel-config.js"; import { renderQrLines } from "../ui/qr.js"; @@ -25,8 +25,9 @@ import { migrateRelayConfigPlan, planRelayConfigMigrationForEnv, type RelayConfi import { createTurnId, deriveSessionLabel, extractImageContent, extractTextContent, getTelegramUserLabel, isAllowedImageMimeType, latestImageFileCandidatesFromText, latestImageFromContent, loadWorkspaceImageFile, sessionKeyOf, summarizeTextDeterministically, toIsoNow } from "../core/utils.js"; import { loadWorkspaceOutboundFile, type RelayOutboundFileKind } from "../core/file-delivery.js"; import { parseMessengerRef } from "../core/messenger-ref.js"; -import { deliverWorkspaceFileToRequester, formatRequesterFileDeliveryResult } from "../core/requester-file-delivery.js"; +import { deliverWorkspaceFileToRequester, formatRequesterFileDeliveryResult, type RelayFileDeliveryRequester } from "../core/requester-file-delivery.js"; import { isStaleExtensionReferenceError, routeUnavailableError, unavailableRouteMessage } from "../core/route-actions.js"; +import { activeSelectionsForSession, bindingSummariesForSession, createPendingSessionHandoff, findPendingSessionHandoff, listPendingSessionHandoffs, normalizeWorkspaceRoot, relayRuntimeInstanceId, registerPendingSessionHandoff, removePendingSessionHandoff, removePendingSessionHandoffsForSession, type PendingSessionHandoff, type SessionHandoffReason } from "../core/session-handoff.js"; import type { ChannelOutboundFile, ChannelRouteAddress } from "../core/channel-adapter.js"; import { TelegramChannelAdapter } from "../adapters/telegram/adapter.js"; import { DEFAULT_DISCORD_MAX_FILE_BYTES } from "../adapters/discord/adapter.js"; @@ -176,6 +177,8 @@ export default function telegramTunnelExtension(pi: ExtensionAPI): void { let configCache: TelegramTunnelConfig | undefined; let runtime: TunnelRuntime | undefined; let telegramRuntimeStatus: { enabled: boolean; started: boolean; error?: string } | undefined; + let latestCommandContext: ExtensionCommandContext | undefined; + let completedHandoff: PendingSessionHandoff | undefined; const discordRuntimes = new Map(); const slackRuntimes = new Map(); const volatileTelegramBindings = new Map(); @@ -912,6 +915,38 @@ export default function telegramTunnelExtension(pi: ExtensionAPI): void { return Boolean(binding && !binding.paused); } + function attachCommandContextToRoute(ctx: ExtensionCommandContext, route = currentRoute): void { + if (!route) return; + try { + const routeWorkspace = route.actions.getWorkspaceRoot?.() ?? route.actions.context.cwd; + if (safeSessionKeyForContext(ctx) !== route.sessionKey || normalizeWorkspaceRoot(ctx.cwd) !== normalizeWorkspaceRoot(routeWorkspace)) return; + } catch { + return; + } + latestCommandContext = ctx; + route.actions.newSession = async (requester) => { + const commandContext = latestCommandContext; + if (commandContext !== ctx) throw routeUnavailableError(); + const handoff = await preparePendingHandoff(route, requester ? "remote-new" : "local-new", requester); + if (requester && !handoff) throw routeUnavailableError("The selected relay binding is no longer active. Use /sessions to choose an online session."); + const parentSession = route.sessionFile; + try { + const result = await commandContext.newSession({ + ...(parentSession ? { parentSession } : {}), + withSession: async (replacementContext) => { + latestContext = replacementContext; + attachCommandContextToRoute(replacementContext); + }, + }); + if (result.cancelled && handoff) removePendingSessionHandoffsForSession(route.sessionKey); + return { cancelled: result.cancelled }; + } catch (error) { + if (handoff) removePendingSessionHandoffsForSession(route.sessionKey); + throw error; + } + }; + } + function buildRoute(ctx: ExtensionContext, binding?: TelegramBindingMetadata, explicitLabel?: string): SessionRoute { const sessionId = ctx.sessionManager.getSessionId(); const sessionFile = ctx.sessionManager.getSessionFile(); @@ -1076,6 +1111,7 @@ export default function telegramTunnelExtension(pi: ExtensionAPI): void { }), }, }; + if (latestCommandContext) attachCommandContextToRoute(latestCommandContext, route); return route; } @@ -1164,7 +1200,15 @@ export default function telegramTunnelExtension(pi: ExtensionAPI): void { } } - async function restoreBinding(ctx: ExtensionContext, config: TelegramTunnelConfig): Promise { + async function restoreBinding(ctx: ExtensionContext, config: TelegramTunnelConfig, preferPersisted = false): Promise { + if (preferPersisted) { + const sessionKey = sessionKeyOf(ctx.sessionManager.getSessionId(), ctx.sessionManager.getSessionFile()); + const migrated = await new TunnelStateStore(config.stateDir).getBindingBySessionKey(sessionKey); + if (migrated?.status === "active") { + volatileTelegramBindings.set(sessionKey, migrated); + return migrated; + } + } let latest: BindingEntryData | undefined; for (const entry of ctx.sessionManager.getBranch()) { if (entry.type === "custom" && (entry.customType === BINDING_ENTRY_TYPE || entry.customType === LEGACY_BINDING_ENTRY_TYPE)) { @@ -1189,6 +1233,41 @@ export default function telegramTunnelExtension(pi: ExtensionAPI): void { return localBinding; } + async function preparePendingHandoff(route: SessionRoute, reason: SessionHandoffReason, remoteRequester?: RelayFileDeliveryRequester): Promise { + const config = configCache; + if (!config) return undefined; + const data = await new TunnelStateStore(config.stateDir).load(); + const bindings = bindingSummariesForSession(data, route.sessionKey); + if (bindings.length === 0) return undefined; + const requester = reason === "remote-new" && remoteRequester + ? bindings.find((binding) => binding.channel === remoteRequester.channel + && binding.instanceId === remoteRequester.instanceId + && binding.conversationId === remoteRequester.conversationId + && binding.userId === remoteRequester.userId) + : undefined; + if (reason === "remote-new" && !requester) return undefined; + const handoff = createPendingSessionHandoff({ + oldSessionKey: route.sessionKey, + oldSessionId: route.sessionId, + oldSessionFile: route.sessionFile, + oldSessionLabel: route.sessionLabel, + runtimeInstanceId: relayRuntimeInstanceId, + machineId: config.machineId ?? "local", + workspaceRoot: route.actions.getWorkspaceRoot?.() ?? route.actions.context.cwd, + reason, + requester, + bindings, + activeSelections: activeSelectionsForSession(data, route.sessionKey), + }); + registerPendingSessionHandoff(handoff, async () => { + const deliveries: Array | undefined> = [notifyTelegramLifecycle(config, route, "offline")]; + for (const discord of discordRuntimes.values()) deliveries.push(discord.notifyLifecycle?.(route, "offline")); + for (const slack of slackRuntimes.values()) deliveries.push(slack.notifyLifecycle?.(route, "offline")); + await Promise.allSettled(deliveries.filter((delivery): delivery is Promise => Boolean(delivery))); + }); + return handoff; + } + async function syncRoute(ctx: ExtensionContext): Promise { latestContext = ctx; let config: TelegramTunnelConfig; @@ -1201,7 +1280,35 @@ export default function telegramTunnelExtension(pi: ExtensionAPI): void { return; } - const restoredBinding = await restoreBinding(ctx, config); + const nextSessionId = ctx.sessionManager.getSessionId(); + const nextSessionFile = ctx.sessionManager.getSessionFile(); + const nextSessionKey = sessionKeyOf(nextSessionId, nextSessionFile); + const handoffMatch = findPendingSessionHandoff({ + sessionKey: nextSessionKey, + runtimeInstanceId: relayRuntimeInstanceId, + machineId: config.machineId ?? "local", + workspaceRoot: ctx.cwd, + }); + completedHandoff = undefined; + if (handoffMatch.kind === "matched") { + const previewRoute = buildRoute(ctx); + const migration = await new TunnelStateStore(config.stateDir).migrateSessionBindings({ + oldSessionKey: handoffMatch.handoff.oldSessionKey, + newSessionKey: nextSessionKey, + newSessionId: nextSessionId, + newSessionFile: nextSessionFile, + newSessionLabel: previewRoute.sessionLabel, + requester: handoffMatch.handoff.reason === "remote-new" && handoffMatch.handoff.requester + ? handoffMatch.handoff.requester + : undefined, + }); + if (migration.telegram.length > 0 || migration.channels.length > 0) { + removePendingSessionHandoff(handoffMatch.handoff.id); + completedHandoff = handoffMatch.handoff; + } + } + + const restoredBinding = await restoreBinding(ctx, config, Boolean(completedHandoff)); const previousSessionKey = currentRoute?.sessionKey; const nextRoute = buildRoute(ctx, restoredBinding); if (nextRoute.sessionKey !== previousSessionKey) clearTurnImageCaches(); @@ -1237,7 +1344,7 @@ export default function telegramTunnelExtension(pi: ExtensionAPI): void { } throw error; }); - await notifyRelayLifecycle(ctx, "online", currentRoute, { telegram: telegramStarted, discordInstances: startedDiscordInstances, slackInstances: startedSlackInstances }); + await notifyRelayLifecycle(ctx, completedHandoff ? "moved" : "online", currentRoute, { telegram: telegramStarted, discordInstances: startedDiscordInstances, slackInstances: startedSlackInstances }); } async function handleSetup(ctx: ExtensionContext): Promise { @@ -1882,6 +1989,7 @@ export default function telegramTunnelExtension(pi: ExtensionAPI): void { if (!currentRoute) await syncRoute(ctx); if (!currentRoute) return; const disconnectedRoute = currentRoute; + removePendingSessionHandoffsForSession(disconnectedRoute.sessionKey); await notifyRelayLifecycle(ctx, "disconnected", disconnectedRoute); currentRoute.binding = undefined; await store.revokeBinding(currentRoute.sessionKey); @@ -1947,6 +2055,7 @@ export default function telegramTunnelExtension(pi: ExtensionAPI): void { }, handler: async (args, ctx) => { latestContext = ctx; + attachCommandContextToRoute(ctx); const intent = parseRelayLocalCommand(args); try { if (intent.unsupportedChannel) { @@ -2052,11 +2161,23 @@ export default function telegramTunnelExtension(pi: ExtensionAPI): void { } }); - pi.on("session_shutdown", async (_event, ctx) => { + pi.on("session_shutdown", async (event, ctx) => { latestContext = ctx; resetToolProgress(); closeConnectQrScreen = undefined; - if (currentRoute) await notifyRelayLifecycle(ctx, "offline", currentRoute); + const shuttingDownRoute = currentRoute; + if (shuttingDownRoute && runtime) await runtime.unregisterRoute(shuttingDownRoute.sessionKey); + if (shuttingDownRoute) { + for (const discordRuntime of discordRuntimes.values()) await discordRuntime.unregisterRoute(shuttingDownRoute.sessionKey); + for (const slackRuntime of slackRuntimes.values()) await slackRuntime.unregisterRoute(shuttingDownRoute.sessionKey); + } + let handoffPending = false; + if (shuttingDownRoute && event.reason === "new") { + const existing = listPendingSessionHandoffs().some((handoff) => handoff.oldSessionKey === shuttingDownRoute.sessionKey); + if (!existing) await preparePendingHandoff(shuttingDownRoute, "local-new").catch(() => undefined); + handoffPending = listPendingSessionHandoffs().some((handoff) => handoff.oldSessionKey === shuttingDownRoute.sessionKey); + } + if (shuttingDownRoute && !handoffPending) await notifyRelayLifecycle(ctx, "offline", shuttingDownRoute); if (currentRoute && configCache) { await new TunnelStateStore(configCache.stateDir).cancelPendingApprovalsForSession(currentRoute.sessionKey, "session shutdown").catch(() => undefined); for (const [approvalId, resolve] of pendingApprovalResolvers) { @@ -2064,13 +2185,7 @@ export default function telegramTunnelExtension(pi: ExtensionAPI): void { pendingApprovalResolvers.delete(approvalId); } } - if (runtime && currentRoute) { - await runtime.unregisterRoute(currentRoute.sessionKey); - } - if (currentRoute) { - for (const discordRuntime of discordRuntimes.values()) await discordRuntime.unregisterRoute(currentRoute.sessionKey); - for (const slackRuntime of slackRuntimes.values()) await slackRuntime.unregisterRoute(currentRoute.sessionKey); - } + latestCommandContext = undefined; }); pi.on("session_tree", async (_event, ctx) => { diff --git a/extensions/relay/state/tunnel-store.ts b/extensions/relay/state/tunnel-store.ts index 0dd27ad..94aeabb 100644 --- a/extensions/relay/state/tunnel-store.ts +++ b/extensions/relay/state/tunnel-store.ts @@ -47,6 +47,22 @@ function parseStoreData(raw: string): TunnelStoreData { }; } +export interface SessionBindingMigrationInput { + oldSessionKey: string; + newSessionKey: string; + newSessionId: string; + newSessionFile?: string; + newSessionLabel: string; + requester?: { channel: ChannelBinding["channel"]; instanceId?: string; conversationId: string; userId: string }; + now?: string; +} + +export interface SessionBindingMigrationResult { + telegram: PersistedBindingRecord[]; + channels: ChannelPersistedBindingRecord[]; + movedSelections: number; +} + export type PendingPairingInspection = | { status: "active"; pairing: PendingPairingRecord } | { status: "missing" | "wrong-channel" | "consumed" | "expired"; pairing?: PendingPairingRecord }; @@ -220,6 +236,75 @@ export class TunnelStateStore { return found; } + async migrateSessionBindings(input: SessionBindingMigrationInput): Promise { + const now = input.now ?? toIsoNow(); + const result: SessionBindingMigrationResult = { telegram: [], channels: [], movedSelections: 0 }; + const movedAddresses = new Map>(); + + await this.update((data) => { + const telegram = data.bindings[input.oldSessionKey]; + const telegramMatchesRequester = !input.requester || input.requester.channel === "telegram" + && input.requester.conversationId === String(telegram?.chatId) + && input.requester.userId === String(telegram?.userId); + if (telegram?.status === "active" && telegramMatchesRequester && !activeTargetTelegramBindingConflicts(data, input, telegram)) { + const moved: PersistedBindingRecord = { + ...telegram, + sessionKey: input.newSessionKey, + sessionId: input.newSessionId, + sessionFile: input.newSessionFile, + sessionLabel: input.newSessionLabel, + lastSeenAt: now, + status: "active", + revokedAt: undefined, + movedAt: undefined, + movedToSessionKey: undefined, + }; + data.bindings[input.oldSessionKey] = { ...telegram, status: "revoked", revokedAt: now, movedAt: now, movedToSessionKey: input.newSessionKey }; + data.bindings[input.newSessionKey] = moved; + result.telegram.push(moved); + movedAddresses.set(channelSelectionStorageKey("telegram", String(telegram.chatId), String(telegram.userId)), { channel: "telegram", conversationId: String(telegram.chatId), userId: String(telegram.userId) }); + } + + for (const [key, binding] of Object.entries(data.channelBindings)) { + if (binding.sessionKey !== input.oldSessionKey || binding.status !== "active") continue; + if (input.requester && (input.requester.channel !== binding.channel + || (input.requester.instanceId ?? "default") !== (binding.instanceId ?? "default") + || input.requester.conversationId !== binding.conversationId + || input.requester.userId !== binding.userId)) continue; + const targetKey = channelBindingStorageKey(binding.channel, input.newSessionKey, binding.instanceId); + const target = data.channelBindings[targetKey]; + if (target && (target.channel !== binding.channel + || (target.instanceId ?? "default") !== (binding.instanceId ?? "default") + || target.conversationId !== binding.conversationId + || target.userId !== binding.userId)) continue; + const moved: ChannelPersistedBindingRecord = { + ...binding, + sessionKey: input.newSessionKey, + sessionId: input.newSessionId, + sessionFile: input.newSessionFile, + sessionLabel: input.newSessionLabel, + lastSeenAt: now, + status: "active", + revokedAt: undefined, + movedAt: undefined, + movedToSessionKey: undefined, + }; + data.channelBindings[key] = { ...binding, status: "revoked", revokedAt: now, movedAt: now, movedToSessionKey: input.newSessionKey }; + data.channelBindings[targetKey] = moved; + result.channels.push(moved); + movedAddresses.set(channelSelectionStorageKey(binding.channel, binding.conversationId, binding.userId), { channel: binding.channel, conversationId: binding.conversationId, userId: binding.userId }); + } + + for (const [key, address] of movedAddresses) { + const selection = data.activeChannelSelections[key]; + if (selection && selection.sessionKey !== input.oldSessionKey && selection.sessionKey !== input.newSessionKey) continue; + data.activeChannelSelections[key] = { ...selection, ...address, sessionKey: input.newSessionKey, updatedAt: now }; + result.movedSelections += 1; + } + }); + return result; + } + async upsertBinding(binding: TelegramBindingMetadata): Promise { const record: PersistedBindingRecord = { ...binding, status: binding.revokedAt ? "revoked" : "active" }; await this.update((data) => { @@ -635,12 +720,17 @@ export class TunnelStateStore { } } +function activeTargetTelegramBindingConflicts(data: TunnelStoreData, input: SessionBindingMigrationInput, source: PersistedBindingRecord): boolean { + const target = data.bindings[input.newSessionKey]; + return Boolean(target && (target.chatId !== source.chatId || target.userId !== source.userId)); +} + function isMissingFileError(error: unknown): boolean { return typeof error === "object" && error !== null && "code" in error && (error as { code?: unknown }).code === "ENOENT"; } function lifecycleStateForNotification(kind: RelayLifecycleEventKind): "online" | "offline" | "disconnected" { - if (kind === "online") return "online"; + if (kind === "online" || kind === "moved") return "online"; if (kind === "offline") return "offline"; return "disconnected"; } diff --git a/openspec/changes/remote-new-session-handoff/tasks.md b/openspec/changes/remote-new-session-handoff/tasks.md index 1596b9a..92e30d8 100644 --- a/openspec/changes/remote-new-session-handoff/tasks.md +++ b/openspec/changes/remote-new-session-handoff/tasks.md @@ -1,52 +1,52 @@ ## 1. Route Action and Command Context -- [ ] 1.1 Add typed new-session route-action outcome types for success, unavailable, busy, unsupported, cancelled, and failure. -- [ ] 1.2 Extend `SessionRouteActions` with a narrow `newSession` action that does not expose raw Pi context to adapters. -- [ ] 1.3 Capture or derive command-capable session-control context only where Pi provides `ExtensionCommandContext` support. -- [ ] 1.4 Return an explicit unsupported-capability outcome when no safe command-capable context is available. -- [ ] 1.5 Add route-action unit tests for success, unavailable, busy, unsupported, cancelled, and stale-route outcomes. +- [x] 1.1 Add typed new-session route-action outcome types for success, unavailable, busy, unsupported, cancelled, and failure. +- [x] 1.2 Extend `SessionRouteActions` with a narrow `newSession` action that does not expose raw Pi context to adapters. +- [x] 1.3 Capture or derive command-capable session-control context only where Pi provides `ExtensionCommandContext` support. +- [x] 1.4 Return an explicit unsupported-capability outcome when no safe command-capable context is available. +- [x] 1.5 Add route-action unit tests for success, unavailable, busy, unsupported, cancelled, and stale-route outcomes. ## 2. Handoff State and Binding Migration -- [ ] 2.1 Add a shared pending-handoff model with old route identity, safe workspace identity, binding summaries, active selections, reason, and TTL. -- [ ] 2.2 Implement strict matching helpers for replacement routes: same runtime/machine, same workspace root, active non-revoked binding, no explicit disconnect, no conflicting new binding, unambiguous candidate. -- [ ] 2.3 Implement binding migration helpers that create new-session bindings, move active selections, and mark old bindings moved/superseded without breaking older state files. -- [ ] 2.4 Ensure old route callbacks/actions/output retrieval become stale after successful handoff. -- [ ] 2.5 Add state-store tests for migration, ambiguity, revoked/paused bindings, moved old bindings, and active selection updates. +- [x] 2.1 Add a shared pending-handoff model with old route identity, safe workspace identity, binding summaries, active selections, reason, and TTL. +- [x] 2.2 Implement strict matching helpers for replacement routes: same runtime/machine, same workspace root, active non-revoked binding, no explicit disconnect, no conflicting new binding, unambiguous candidate. +- [x] 2.3 Implement binding migration helpers that create new-session bindings, move active selections, and mark old bindings moved/superseded without breaking older state files. +- [x] 2.4 Ensure old route callbacks/actions/output retrieval become stale after successful handoff. +- [x] 2.5 Add state-store tests for migration, ambiguity, revoked/paused bindings, moved old bindings, and active selection updates. ## 3. Local Session Renewal Lifecycle -- [ ] 3.1 On `session_shutdown`, unregister the old route immediately but create a bounded pending handoff when active bindings make renewal possible. -- [ ] 3.2 Delay offline lifecycle notification during the handoff window and flush it when the handoff expires or fails. -- [ ] 3.3 On `session_start`, migrate eligible bindings to the replacement route and send a safe moved-to-new-session notification. -- [ ] 3.4 Preserve normal offline notification behavior for true shutdowns, explicit disconnects, and unsafe/ambiguous handoffs. -- [ ] 3.5 Add runtime lifecycle tests for local `/new` success, expiry, ambiguous candidate, explicit disconnect, and stale old actions. +- [x] 3.1 On `session_shutdown`, unregister the old route immediately but create a bounded pending handoff when active bindings make renewal possible. +- [x] 3.2 Delay offline lifecycle notification during the handoff window and flush it when the handoff expires or fails. +- [x] 3.3 On `session_start`, migrate eligible bindings to the replacement route and send a safe moved-to-new-session notification. +- [x] 3.4 Preserve normal offline notification behavior for true shutdowns, explicit disconnects, and unsafe/ambiguous handoffs. +- [x] 3.5 Add runtime lifecycle tests for local `/new` success, expiry, ambiguous candidate, explicit disconnect, and stale old actions. ## 4. Remote New-Session Command -- [ ] 4.1 Add canonical `/new` command parsing and help text for Telegram, with shared command definition metadata where applicable. -- [ ] 4.2 Route remote `/new` through authorization, selected-route resolution, idle/busy checks, and the new-session route action. -- [ ] 4.3 Implement safe success, offline, ambiguous, busy, unsupported, cancelled, and failure response text. -- [ ] 4.4 Decide and implement initial busy policy, defaulting to refusal with guidance rather than replacing an active turn. -- [ ] 4.5 Add Telegram direct runtime tests for authorized, unauthorized, offline, ambiguous, busy, unsupported, cancelled, and successful remote `/new`. +- [x] 4.1 Add canonical `/new` command parsing and help text for Telegram, with shared command definition metadata where applicable. +- [x] 4.2 Route remote `/new` through authorization, selected-route resolution, idle/busy checks, and the new-session route action. +- [x] 4.3 Implement safe success, offline, ambiguous, busy, unsupported, cancelled, and failure response text. +- [x] 4.4 Decide and implement initial busy policy, defaulting to refusal with guidance rather than replacing an active turn. +- [x] 4.5 Add Telegram direct runtime tests for authorized, unauthorized, offline, ambiguous, busy, unsupported, cancelled, and successful remote `/new`. ## 5. Broker and Adapter Parity -- [ ] 5.1 Extend broker client protocol to request new-session route actions and report typed outcomes without leaking secrets. -- [ ] 5.2 Implement Telegram broker `/new` handling with equivalent direct-runtime behavior and active selection migration. -- [ ] 5.3 Update Slack and Discord command surfaces to execute shared new-session behavior or return explicit capability fallback. -- [ ] 5.4 Ensure paused, revoked, moved, state-unavailable, and destination-mismatch binding checks suppress new-session side effects. -- [ ] 5.5 Add broker, Slack, and Discord parity tests for command routing, outcomes, and fallback behavior. +- [x] 5.1 Extend broker client protocol to request new-session route actions and report typed outcomes without leaking secrets. +- [x] 5.2 Implement Telegram broker `/new` handling with equivalent direct-runtime behavior and active selection migration. +- [x] 5.3 Update Slack and Discord command surfaces to execute shared new-session behavior or return explicit capability fallback. +- [x] 5.4 Ensure paused, revoked, moved, state-unavailable, and destination-mismatch binding checks suppress new-session side effects. +- [x] 5.5 Add broker, Slack, and Discord parity tests for command routing, outcomes, and fallback behavior. ## 6. Documentation and UX -- [ ] 6.1 Update README/help text to document `/new`, busy/offline limitations, and handoff behavior. -- [ ] 6.2 Add safe lifecycle notification copy for successful session handoff and failed/expired handoff guidance. -- [ ] 6.3 Document that remote `/new` starts a replacement session for an online selected route and does not start Pi when the route is offline. -- [ ] 6.4 Add smoke-test guidance for local `/new` handoff and remote `/new` through Telegram. +- [x] 6.1 Update README/help text to document `/new`, busy/offline limitations, and handoff behavior. +- [x] 6.2 Add safe lifecycle notification copy for successful session handoff and failed/expired handoff guidance. +- [x] 6.3 Document that remote `/new` starts a replacement session for an online selected route and does not start Pi when the route is offline. +- [x] 6.4 Add smoke-test guidance for local `/new` handoff and remote `/new` through Telegram. ## 7. Validation -- [ ] 7.1 Run `npm run typecheck`. -- [ ] 7.2 Run `npm test`. -- [ ] 7.3 Run `openspec validate remote-new-session-handoff --strict`. +- [x] 7.1 Run `npm run typecheck`. +- [x] 7.2 Run `npm test`. +- [x] 7.3 Run `openspec validate remote-new-session-handoff --strict`. diff --git a/skills/relay/SKILL.md b/skills/relay/SKILL.md index 966e736..8059e98 100644 --- a/skills/relay/SKILL.md +++ b/skills/relay/SKILL.md @@ -36,6 +36,7 @@ Use the extension command for all runtime actions: - `/recent` or `/activity` - `/forget ` - `/to ` +- `/new` - `/summary` - `/full` - `/images` @@ -53,6 +54,8 @@ Plain text messages are delivered as normal Pi prompts when idle and as follow-u If multiple Pi sessions are paired to the same Telegram chat, use `/sessions` to list numbered sessions with stable visual markers, aliases/labels, online/offline state, active marker, idle/busy state, model, last activity, and quick-action buttons. Multi-session notifications include the same marker + label so source sessions are easier to distinguish. Use `/use ` to switch the active one, `/forget ` to remove an offline paired session from the list, or `/to ` to send a one-shot prompt without changing the active session. Quote `/to` labels that contain spaces, for example `/to "docs team" run tests`. Pair with `/relay connect telegram docs` or run `/use docs` then `/alias phone` when you want friendlier names; otherwise PiRelay falls back to the Pi session name, project folder name, session file basename, then a short session id. +Use `/new` only for the selected online idle route. It starts a replacement Pi session and safely moves eligible relay control to it; it does not start an offline Pi process, and busy or unsupported routes refuse with guidance. + During long-running turns, PiRelay sends safe, rate-limited progress updates by default. Use `/progress quiet` for completion-only style behavior, `/progress verbose` for more frequent safe updates, and `/recent` to retrieve recent safe progress/lifecycle activity on demand. Approval gates are opt-in. When `approvalGates.enabled` and explicit rules are configured, sensitive remote tool calls pause and ask the active authorized requester to approve once, approve for session, or deny. Text fallback commands are `relay approval approve `, `relay approval approve-session `, and `relay approval deny `. Timeout, stale actions, revoked/paused bindings, or missing requester context block safely. diff --git a/tests/discord-runtime.test.ts b/tests/discord-runtime.test.ts index 4142156..0ef8e74 100644 --- a/tests/discord-runtime.test.ts +++ b/tests/discord-runtime.test.ts @@ -1588,6 +1588,8 @@ describe("DiscordRuntime", () => { session.notification.recentActivity = [{ id: "p1", kind: "tool", text: "Ran tests", at: Date.now() }]; session.notification.latestImages = { turnId: "turn-1", count: 1, skipped: 0, contentCount: 1 }; session.actions.getLatestImages = async () => [{ id: "img1", turnId: "turn-1", fileName: "render.png", mimeType: "image/png", data: Buffer.from([1, 2, 3]).toString("base64"), byteSize: 3 }]; + const newSession = vi.fn(async () => ({ cancelled: false })); + session.actions.newSession = newSession; session.actions.getImageByPath = async (path) => path === "outputs/render.png" ? { ok: true, image: { id: "img2", turnId: "turn-2", fileName: "render-path.png", mimeType: "image/png", data: Buffer.from([4, 5, 6]).toString("base64"), byteSize: 3 } } : { ok: false, error: "Image file not found." }; @@ -1607,8 +1609,12 @@ describe("DiscordRuntime", () => { identity: { displayName: "zikolach" }, }); - for (const command of ["relay help", "relay status", "relay sessions", "relay full", "relay summary", "relay recent", "relay progress", "relay alias phone", "relay progress verbose", "relay images", "relay send-image outputs/render.png", "relay send-file report.md Report", "relay steer go", "relay followup next", "relay to phone one shot", "relay use phone", "relay forget missing", "relay abort", "relay compact", "relay pause", "relay resume", "relay disconnect"] as const) { + for (const command of ["relay help", "relay status", "relay sessions", "relay full", "relay summary", "relay recent", "relay progress", "relay alias phone", "relay progress verbose", "relay images", "relay send-image outputs/render.png", "relay send-file report.md Report", "relay steer go", "relay followup next", "relay to phone one shot", "relay use phone", "relay forget missing", "relay abort", "relay compact", "relay new", "relay pause", "relay resume", "relay disconnect"] as const) { await ops.handler?.(discordMessage(command)); + if (command === "relay new") { + session.actions.newSession = undefined; + await ops.handler?.(discordMessage("relay new")); + } } const text = ops.messages.map((message) => message.content).join("\n"); @@ -1629,6 +1635,9 @@ describe("DiscordRuntime", () => { expect(sendUserMessage).toHaveBeenCalledWith("go", undefined); expect(sendUserMessage).toHaveBeenCalledWith("next", undefined); expect(sendUserMessage).toHaveBeenCalledWith("one shot", undefined); + expect(newSession).toHaveBeenCalledOnce(); + expect(text).toContain("New Pi session started"); + expect(text).toContain("cannot start a new session remotely"); expect(ops.files.map((file) => file.fileName)).toEqual(["render.png", "render-path.png", "report.md"]); expect(ops.files.at(-1)).toMatchObject({ caption: "Report", mimeType: "text/markdown" }); }); diff --git a/tests/integration.test.ts b/tests/integration.test.ts index d60d05c..87a4546 100644 --- a/tests/integration.test.ts +++ b/tests/integration.test.ts @@ -758,8 +758,146 @@ describe("PiRelay integration behavior", () => { await pi.emit("session_start", {}, context); expect(fakeRuntime.sendToBoundChat).toHaveBeenCalledWith(binding.sessionKey, expect.stringContaining("back online")); + const handoffs = await import("../extensions/relay/core/session-handoff.js"); + handoffs.registerPendingSessionHandoff(handoffs.createPendingSessionHandoff({ + oldSessionKey: binding.sessionKey, oldSessionId: binding.sessionId, oldSessionFile: binding.sessionFile, oldSessionLabel: binding.sessionLabel, + runtimeInstanceId: handoffs.relayRuntimeInstanceId, machineId: config.machineId ?? "local", workspaceRoot: process.cwd(), reason: "local-new", + bindings: [{ channel: "telegram", instanceId: "default", conversationId: String(binding.chatId), userId: String(binding.userId), paused: false }], activeSelections: [], + })); await pi.runCommand("relay", "disconnect", context); expect(fakeRuntime.sendToBoundChat).toHaveBeenCalledWith(binding.sessionKey, expect.stringContaining("disconnected locally")); + expect(handoffs.listPendingSessionHandoffs().filter((pending) => pending.oldSessionKey === binding.sessionKey)).toEqual([]); + }); + + it("exposes remote new-session capability only after a command-capable context is captured", async () => { + const config = await createRuntimeConfig("pi-new-command-context-"); + vi.stubEnv("TELEGRAM_BOT_TOKEN", config.botToken); + vi.stubEnv("PI_TELEGRAM_TUNNEL_STATE_DIR", config.stateDir); + const binding = createBinding("new-command-context", 557, 44); + await new TunnelStateStore(config.stateDir).upsertBinding(binding); + const registerRoute = vi.fn(async (_route: SessionRoute) => undefined); + const fakeRuntime: TunnelRuntime = { + setup: undefined, + start: vi.fn(async () => undefined), stop: vi.fn(async () => undefined), restartBrokerProcess: vi.fn(async () => undefined), + ensureSetup: vi.fn(async () => ({ botId: 123456, botUsername: "pi_test_bot", botDisplayName: "Pi Test Bot", validatedAt: new Date().toISOString() })), + registerRoute, unregisterRoute: vi.fn(async () => undefined), getStatus: vi.fn(() => undefined), sendToBoundChat: vi.fn(async () => undefined), + }; + vi.doMock("../extensions/relay/adapters/telegram/runtime.js", () => ({ getOrCreateTunnelRuntime: () => fakeRuntime, sendSessionNotification: vi.fn(async () => undefined) })); + const { default: relayExtension } = await import("../extensions/relay/index.js"); + const pi = createMockPi(); + const { context } = createMockContext("new-command-context"); + context.newSession = vi.fn(async () => ({ cancelled: false })); + relayExtension(pi.api as any); + await pi.emit("session_start", { reason: "startup" }, context); + const route = registerRoute.mock.calls.at(-1)?.[0] as SessionRoute; + expect(route.actions.newSession).toBeUndefined(); + route.actions.getWorkspaceRoot = undefined; + + await pi.runCommand("relay", "status", context); + expect(route.actions.newSession).toBeTypeOf("function"); + await expect(route.actions.newSession?.()).resolves.toEqual({ cancelled: false }); + expect(context.newSession).toHaveBeenCalledWith(expect.objectContaining({ parentSession: binding.sessionFile, withSession: expect.any(Function) })); + }); + + it("hands an active binding to a local replacement session without a false offline notification", async () => { + const config = await createRuntimeConfig("pi-lifecycle-handoff-"); + vi.stubEnv("TELEGRAM_BOT_TOKEN", config.botToken); + vi.stubEnv("PI_TELEGRAM_TUNNEL_STATE_DIR", config.stateDir); + const oldBinding = createBinding("handoff-old", 556, 43); + await new TunnelStateStore(config.stateDir).upsertBinding(oldBinding); + + const fakeRuntime: TunnelRuntime = { + setup: undefined, + start: vi.fn(async () => undefined), + stop: vi.fn(async () => undefined), + ensureSetup: vi.fn(async () => ({ botId: 123456, botUsername: "pi_test_bot", botDisplayName: "Pi Test Bot", validatedAt: new Date().toISOString() })), + registerRoute: vi.fn(async () => undefined), + unregisterRoute: vi.fn(async () => undefined), + getStatus: vi.fn(() => undefined), + sendToBoundChat: vi.fn(async () => undefined), + }; + vi.doMock("../extensions/relay/adapters/telegram/runtime.js", () => ({ + getOrCreateTunnelRuntime: () => fakeRuntime, + sendSessionNotification: vi.fn(async () => undefined), + })); + + const { default: relayExtension } = await import("../extensions/relay/index.js"); + const pi = createMockPi(); + const oldContext = createMockContext("handoff-old").context; + relayExtension(pi.api as any); + await pi.emit("session_start", { reason: "startup" }, oldContext); + await pi.runCommand("relay", "status", oldContext); + const oldRoute = (fakeRuntime.registerRoute as ReturnType).mock.calls.at(-1)?.[0] as SessionRoute; + expect(oldRoute.actions.newSession).toBeTypeOf("function"); + await pi.emit("session_shutdown", { reason: "new" }, oldContext); + + expect(fakeRuntime.unregisterRoute).toHaveBeenCalledWith(oldBinding.sessionKey); + expect(fakeRuntime.sendToBoundChat).not.toHaveBeenCalledWith(oldBinding.sessionKey, expect.stringContaining("went offline")); + + const newContext = createMockContext("handoff-new").context; + await pi.emit("session_start", { reason: "new", previousSessionFile: oldBinding.sessionFile }, newContext); + const newKey = sessionKeyOf("handoff-new", "/tmp/handoff-new.jsonl"); + const store = new TunnelStateStore(config.stateDir); + expect(await store.getBindingBySessionKey(oldBinding.sessionKey)).toMatchObject({ status: "revoked", movedToSessionKey: newKey }); + expect(await store.getBindingBySessionKey(newKey)).toMatchObject({ status: "active", chatId: 556, userId: 43 }); + expect(fakeRuntime.sendToBoundChat).toHaveBeenCalledWith(newKey, expect.stringContaining("Relay control moved here automatically")); + await expect(oldRoute.actions.newSession?.()).rejects.toThrow("unavailable"); + }); + + it("flushes the offline lifecycle notification when a new-session handoff expires", async () => { + const config = await createRuntimeConfig("pi-lifecycle-handoff-expiry-"); + vi.stubEnv("TELEGRAM_BOT_TOKEN", config.botToken); + vi.stubEnv("PI_TELEGRAM_TUNNEL_STATE_DIR", config.stateDir); + const binding = createBinding("handoff-expiry", 558, 45); + await new TunnelStateStore(config.stateDir).upsertBinding(binding); + const fakeRuntime: TunnelRuntime = { + setup: undefined, start: vi.fn(async () => undefined), stop: vi.fn(async () => undefined), ensureSetup: vi.fn(async () => ({ botId: 1, botUsername: "bot", botDisplayName: "Bot", validatedAt: new Date().toISOString() })), + registerRoute: vi.fn(async () => undefined), unregisterRoute: vi.fn(async () => undefined), getStatus: vi.fn(() => undefined), sendToBoundChat: vi.fn(async () => undefined), + }; + vi.doMock("../extensions/relay/adapters/telegram/runtime.js", () => ({ getOrCreateTunnelRuntime: () => fakeRuntime, sendSessionNotification: vi.fn(async () => undefined) })); + const { default: relayExtension } = await import("../extensions/relay/index.js"); + const pi = createMockPi(); + const { context } = createMockContext("handoff-expiry"); + relayExtension(pi.api as any); + await pi.emit("session_start", { reason: "startup" }, context); + vi.useFakeTimers(); + await pi.emit("session_shutdown", { reason: "new" }, context); + expect(fakeRuntime.sendToBoundChat).not.toHaveBeenCalledWith(binding.sessionKey, expect.stringContaining("went offline")); + vi.setSystemTime(Date.now() + 5_000); + await (await import("../extensions/relay/core/session-handoff.js")).expirePendingSessionHandoffs(); + expect(fakeRuntime.sendToBoundChat).toHaveBeenCalledWith(binding.sessionKey, expect.stringContaining("went offline locally")); + }); + + it("fails a same-workspace ambiguous handoff closed without migrating bindings", async () => { + const config = await createRuntimeConfig("pi-lifecycle-handoff-ambiguous-"); + vi.stubEnv("TELEGRAM_BOT_TOKEN", config.botToken); + vi.stubEnv("PI_TELEGRAM_TUNNEL_STATE_DIR", config.stateDir); + const first = createBinding("handoff-ambiguous-a", 559, 46); + const second = createBinding("handoff-ambiguous-b", 560, 47); + const store = new TunnelStateStore(config.stateDir); + await store.upsertBinding(first); + await store.upsertBinding(second); + const fakeRuntime: TunnelRuntime = { + setup: undefined, start: vi.fn(async () => undefined), stop: vi.fn(async () => undefined), ensureSetup: vi.fn(async () => ({ botId: 1, botUsername: "bot", botDisplayName: "Bot", validatedAt: new Date().toISOString() })), + registerRoute: vi.fn(async () => undefined), unregisterRoute: vi.fn(async () => undefined), getStatus: vi.fn(() => undefined), sendToBoundChat: vi.fn(async () => undefined), + }; + vi.doMock("../extensions/relay/adapters/telegram/runtime.js", () => ({ getOrCreateTunnelRuntime: () => fakeRuntime, sendSessionNotification: vi.fn(async () => undefined) })); + const handoffs = await import("../extensions/relay/core/session-handoff.js"); + for (const binding of [first, second]) { + handoffs.registerPendingSessionHandoff(handoffs.createPendingSessionHandoff({ + oldSessionKey: binding.sessionKey, oldSessionId: binding.sessionId, oldSessionFile: binding.sessionFile, oldSessionLabel: binding.sessionLabel, + runtimeInstanceId: handoffs.relayRuntimeInstanceId, machineId: config.machineId ?? "local", workspaceRoot: process.cwd(), reason: "local-new", + bindings: [{ channel: "telegram", instanceId: "default", conversationId: String(binding.chatId), userId: String(binding.userId), paused: false }], activeSelections: [], + })); + } + const { default: relayExtension } = await import("../extensions/relay/index.js"); + const pi = createMockPi(); + relayExtension(pi.api as any); + await pi.emit("session_start", { reason: "new" }, createMockContext("handoff-ambiguous-new").context); + expect(await store.getBindingBySessionKey(first.sessionKey)).toMatchObject({ status: "active" }); + expect(await store.getBindingBySessionKey(second.sessionKey)).toMatchObject({ status: "active" }); + expect(await store.getBindingBySessionKey(sessionKeyOf("handoff-ambiguous-new", "/tmp/handoff-ambiguous-new.jsonl"))).toBeUndefined(); + for (const pending of handoffs.listPendingSessionHandoffs()) handoffs.removePendingSessionHandoff(pending.id); }); it("does not mark restored binding online when Telegram registration fails", async () => { @@ -3495,6 +3633,26 @@ describe("PiRelay integration behavior", () => { await waitFor(() => clientResponses.some((message) => message.requestId === "broker-missing-pipeline-version")); expect(clientResponses.find((message) => message.requestId === "broker-missing-pipeline-version")).toMatchObject({ ok: false, error: "Invalid relay pipeline protocol version." }); + setIdle(true); + const newSession = vi.fn(async () => ({ cancelled: false })); + route.actions.newSession = newSession; + sockets[0]!.write(`${JSON.stringify({ + type: "request", + requestId: "broker-new-session-1", + action: "newSession", + sessionKey: route.sessionKey, + requester: { channel: "telegram", instanceId: "default", conversationId: "8001", userId: "9001", sessionKey: route.sessionKey, safeLabel: "Telegram owner", createdAt: Date.now() }, + })}\n`); + await waitFor(() => clientResponses.some((message) => message.requestId === "broker-new-session-1")); + expect(clientResponses.find((message) => message.requestId === "broker-new-session-1")).toMatchObject({ ok: true, result: { kind: "success", result: { started: true } } }); + expect(newSession).toHaveBeenCalledWith(expect.objectContaining({ channel: "telegram", conversationId: "8001", userId: "9001" })); + + route.actions.newSession = undefined; + sockets[0]!.write(`${JSON.stringify({ type: "request", requestId: "broker-new-session-unsupported", action: "newSession", sessionKey: route.sessionKey })}\n`); + await waitFor(() => clientResponses.some((message) => message.requestId === "broker-new-session-unsupported")); + expect(clientResponses.find((message) => message.requestId === "broker-new-session-unsupported")).toMatchObject({ ok: true, result: { kind: "unsupported" } }); + + setIdle(false); const multimodalContent = [ { type: "text", text: "look at this" }, { type: "image", data: Buffer.from("img").toString("base64"), mimeType: "image/png" }, diff --git a/tests/relay/binding-authority.test.ts b/tests/relay/binding-authority.test.ts index b1834df..785d204 100644 --- a/tests/relay/binding-authority.test.ts +++ b/tests/relay/binding-authority.test.ts @@ -61,6 +61,7 @@ describe("binding authority", () => { ["active", telegramBinding(), { chatId: 100, userId: 200 }, "active"], ["paused", telegramBinding({ paused: true }), { chatId: 100, userId: 200 }, "paused"], ["revoked", telegramBinding({ status: "revoked", revokedAt: "2026-05-15T00:01:00.000Z" }), { chatId: 100, userId: 200 }, "revoked"], + ["handoff-moved", telegramBinding({ status: "revoked", revokedAt: "2026-05-15T00:01:00.000Z", movedToSessionKey: "session-2" }), { chatId: 100, userId: 200 }, "moved"], ["moved-chat", telegramBinding({ chatId: 999 }), { chatId: 100, userId: 200 }, "moved"], ["moved-user", telegramBinding({ userId: 999 }), { chatId: 100, userId: 200 }, "moved"], ] as const)("classifies Telegram %s bindings", (_name, binding, expected, kind) => { @@ -81,6 +82,7 @@ describe("binding authority", () => { ["active", channelBinding(), { conversationId: "C1", userId: "U1" }, "active"], ["paused", channelBinding({ paused: true }), { conversationId: "C1", userId: "U1" }, "paused"], ["revoked", channelBinding({ status: "revoked", revokedAt: "2026-05-15T00:01:00.000Z" }), { conversationId: "C1", userId: "U1" }, "revoked"], + ["handoff-moved", channelBinding({ status: "revoked", revokedAt: "2026-05-15T00:01:00.000Z", movedToSessionKey: "session-2" }), { conversationId: "C1", userId: "U1" }, "moved"], ["moved-conversation", channelBinding({ conversationId: "C2" }), { conversationId: "C1", userId: "U1" }, "moved"], ["moved-user", channelBinding({ userId: "U2" }), { conversationId: "C1", userId: "U1" }, "moved"], ] as const)("classifies channel %s bindings", (_name, binding, expected, kind) => { diff --git a/tests/relay/route-actions.test.ts b/tests/relay/route-actions.test.ts index e700b32..e5de4b8 100644 --- a/tests/relay/route-actions.test.ts +++ b/tests/relay/route-actions.test.ts @@ -11,6 +11,7 @@ import { compactRouteSafely, deliverRoutePrompt, latestRouteImagesSafely, + newSessionRouteSafely, routeActionOutcomeFromError, routeActionSuccess, routeActionUnavailable, @@ -116,6 +117,16 @@ describe("route action lifetime helpers", () => { expect(getImageByPath).toHaveBeenCalledWith("missing.png"); }); + it("handles typed new-session outcomes without using raw context", async () => { + await expect(newSessionRouteSafely(route({ newSession: async () => ({ cancelled: false }) }))).resolves.toEqual({ kind: "success", result: { started: true } }); + await expect(newSessionRouteSafely(route({ isIdle: () => undefined, newSession: async () => ({ cancelled: false }) }))).resolves.toEqual({ kind: "unavailable", message: unavailableRouteMessage() }); + await expect(newSessionRouteSafely(route({ isIdle: () => false, newSession: async () => ({ cancelled: false }) }))).resolves.toMatchObject({ kind: "busy" }); + await expect(newSessionRouteSafely(route())).resolves.toMatchObject({ kind: "unsupported" }); + await expect(newSessionRouteSafely(route({ newSession: async () => ({ cancelled: true }) }))).resolves.toMatchObject({ kind: "cancelled" }); + await expect(newSessionRouteSafely(route({ newSession: async () => { throw new Error(STALE_EXTENSION_ERROR); } }))).resolves.toEqual({ kind: "unavailable", message: unavailableRouteMessage() }); + await expect(newSessionRouteSafely(route({ newSession: async () => { throw new Error("boom"); } }))).resolves.toMatchObject({ kind: "failed", safeMessage: "Could not start a new Pi session." }); + }); + it("handles abort outcomes and rollback", () => { const abort = vi.fn(); const busy = route({ isIdle: () => false, abort }); diff --git a/tests/relay/session-handoff.test.ts b/tests/relay/session-handoff.test.ts new file mode 100644 index 0000000..b52cfe5 --- /dev/null +++ b/tests/relay/session-handoff.test.ts @@ -0,0 +1,79 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { + createPendingSessionHandoff, + matchPendingSessionHandoffs, + registerPendingSessionHandoff, + removePendingSessionHandoff, + takePendingSessionHandoff, + type PendingSessionHandoff, +} from "../../extensions/relay/core/session-handoff.js"; + +function handoff(overrides: Partial = {}): PendingSessionHandoff { + return createPendingSessionHandoff({ + oldSessionKey: "old", + oldSessionId: "old-id", + oldSessionLabel: "old", + runtimeInstanceId: "runtime-1", + machineId: "machine-1", + workspaceRoot: "/workspace", + reason: "local-new", + bindings: [{ channel: "telegram", instanceId: "default", conversationId: "1", userId: "2", paused: false }], + activeSelections: [], + now: 1_000, + ttlMs: 5_000, + ...overrides, + }); +} + +const replacement = { sessionKey: "new", runtimeInstanceId: "runtime-1", machineId: "machine-1", workspaceRoot: "/workspace" }; + +afterEach(() => { + vi.useRealTimers(); +}); + +describe("session handoff matching", () => { + it("matches only the same runtime, machine, workspace, and live TTL", () => { + expect(matchPendingSessionHandoffs([handoff()], replacement, 2_000)).toMatchObject({ kind: "matched", handoff: { oldSessionKey: "old" } }); + expect(matchPendingSessionHandoffs([handoff({ machineId: "other" })], replacement, 2_000)).toEqual({ kind: "none" }); + expect(matchPendingSessionHandoffs([handoff({ workspaceRoot: "/other" })], replacement, 2_000)).toEqual({ kind: "none" }); + expect(matchPendingSessionHandoffs([handoff({ explicitDisconnect: true })], replacement, 2_000)).toEqual({ kind: "none" }); + expect(matchPendingSessionHandoffs([handoff()], replacement, 6_000)).toEqual({ kind: "none" }); + }); + + it("fails closed when multiple pending handoffs match", () => { + expect(matchPendingSessionHandoffs([handoff(), handoff({ oldSessionKey: "old-2", oldSessionId: "old-2" })], replacement, 2_000)).toMatchObject({ kind: "ambiguous", candidates: [{ oldSessionKey: "old" }, { oldSessionKey: "old-2" }] }); + }); + + it("runs expiry behavior when synchronous pruning wins the timer race", async () => { + vi.useFakeTimers(); + vi.setSystemTime(1_000); + const pending = handoff(); + const expired = vi.fn(); + registerPendingSessionHandoff(pending, expired); + + vi.setSystemTime(6_000); + expect(takePendingSessionHandoff(replacement)).toEqual({ kind: "none" }); + await vi.runAllTicks(); + expect(expired).toHaveBeenCalledOnce(); + expect(expired).toHaveBeenCalledWith(pending); + await vi.runAllTimersAsync(); + expect(expired).toHaveBeenCalledOnce(); + }); + + it("takes a matched handoff once and expires bounded records", async () => { + vi.useFakeTimers(); + vi.setSystemTime(1_000); + const pending = handoff(); + const expired = vi.fn(); + registerPendingSessionHandoff(pending, expired); + expect(takePendingSessionHandoff(replacement, 2_000)).toMatchObject({ kind: "matched" }); + expect(takePendingSessionHandoff(replacement, 2_000)).toEqual({ kind: "none" }); + expect(expired).not.toHaveBeenCalled(); + + const expiring = handoff({ oldSessionKey: "old-expiring", oldSessionId: "old-expiring" }); + registerPendingSessionHandoff(expiring, expired); + await vi.advanceTimersByTimeAsync(5_000); + expect(expired).toHaveBeenCalledWith(expiring); + removePendingSessionHandoff(expiring.id); + }); +}); diff --git a/tests/runtime.test.ts b/tests/runtime.test.ts index f27f2a6..6255c80 100644 --- a/tests/runtime.test.ts +++ b/tests/runtime.test.ts @@ -2357,6 +2357,70 @@ describe("InProcessTunnelRuntime", () => { expect(sent).toContain("The Pi session is unavailable. Resume it locally, then try again."); }); + it("routes authorized Telegram /new through typed route outcomes", async () => { + const config = await createRuntimeConfig(); + const store = new TunnelStateStore(config.stateDir); + const runtime = new InProcessTunnelRuntime(config, store); + const binding: TelegramBindingMetadata = { + sessionKey: "session-new:/tmp/session-new.jsonl", + sessionId: "session-new", + sessionFile: "/tmp/session-new.jsonl", + sessionLabel: "new.jsonl", + chatId: 1017, + userId: 37, + username: "owner", + boundAt: new Date().toISOString(), + lastSeenAt: new Date().toISOString(), + }; + const { route, setIdle } = createRoute(binding, true); + const newSession = vi.fn(async () => ({ cancelled: false })); + route.actions.newSession = newSession; + await store.upsertBinding(binding); + (runtime as any).routes.set(route.sessionKey, route); + const sent: string[] = []; + (runtime as any).api = { sendPlainText: async (_chatId: number, text: string) => sent.push(text) }; + + await (runtime as any).processInbound({ updateId: 46, messageId: 46, text: "/new", chat: { id: 1017, type: "private" }, user: { id: 37, username: "owner" } }); + expect(newSession).toHaveBeenCalledWith(expect.objectContaining({ channel: "telegram", conversationId: "1017", userId: "37" })); + expect(sent.at(-1)).toContain("New Pi session started"); + + setIdle(false); + await (runtime as any).processInbound({ updateId: 47, messageId: 47, text: "/new", chat: { id: 1017, type: "private" }, user: { id: 37, username: "owner" } }); + expect(newSession).toHaveBeenCalledOnce(); + expect(sent.at(-1)).toContain("Pi is busy"); + + await (runtime as any).processInbound({ updateId: 48, messageId: 48, text: "/new", chat: { id: 1017, type: "private" }, user: { id: 99, username: "stranger" } }); + expect(newSession).toHaveBeenCalledOnce(); + expect(sent.at(-1)).toContain("Unauthorized"); + }); + + it("renders unsupported, cancelled, failed, and unavailable Telegram /new outcomes", async () => { + const config = await createRuntimeConfig(); + const store = new TunnelStateStore(config.stateDir); + const runtime = new InProcessTunnelRuntime(config, store); + const binding: TelegramBindingMetadata = { + sessionKey: "session-new-outcomes:/tmp/session-new-outcomes.jsonl", sessionId: "session-new-outcomes", sessionFile: "/tmp/session-new-outcomes.jsonl", sessionLabel: "outcomes.jsonl", chatId: 1018, userId: 38, boundAt: new Date().toISOString(), lastSeenAt: new Date().toISOString(), + }; + const { route } = createRoute(binding, true); + await store.upsertBinding(binding); + (runtime as any).routes.set(route.sessionKey, route); + const sent: string[] = []; + (runtime as any).api = { sendPlainText: async (_chatId: number, text: string) => sent.push(text) }; + const message = { updateId: 49, messageId: 49, text: "/new", chat: { id: 1018, type: "private" }, user: { id: 38, username: "owner" } }; + + await (runtime as any).processInbound(message); + expect(sent.at(-1)).toContain("cannot start a new session remotely"); + route.actions.newSession = async () => ({ cancelled: true }); + await (runtime as any).processInbound({ ...message, updateId: 50 }); + expect(sent.at(-1)).toContain("cancelled locally"); + route.actions.newSession = async () => { throw new Error("secret internal failure"); }; + await (runtime as any).processInbound({ ...message, updateId: 51 }); + expect(sent.at(-1)).toBe("Could not start a new Pi session."); + route.actions.isIdle = () => undefined; + await (runtime as any).processInbound({ ...message, updateId: 52 }); + expect(sent.at(-1)).toContain("unavailable"); + }); + it("supports text /use, /to, and /forget session controls in-process", async () => { const config = await createRuntimeConfig(); const store = new TunnelStateStore(config.stateDir); diff --git a/tests/slack-runtime.test.ts b/tests/slack-runtime.test.ts index bdeb678..f2d837f 100644 --- a/tests/slack-runtime.test.ts +++ b/tests/slack-runtime.test.ts @@ -649,6 +649,8 @@ describe("SlackRuntime foundations", () => { const testRoute = route(); testRoute.notification.lastAssistantText = "full output"; testRoute.notification.lastSummary = "summary output"; + const newSession = vi.fn(async () => ({ cancelled: false })); + testRoute.actions.newSession = newSession; const store = new TunnelStateStore(runtimeConfig.stateDir); await store.upsertChannelBinding({ channel: "slack", @@ -709,6 +711,12 @@ describe("SlackRuntime foundations", () => { await send("/to Docs hello there", "45"); expect(testRoute.actions.sendUserMessage).toHaveBeenCalledWith("hello there", undefined); expect(operations.ephemeral.at(-1)).toMatchObject({ channel: "D1", user: "U_DRIVER", text: "Pi is working…", threadTs: "45" }); + await send("/new", "45.5"); + expect(newSession).toHaveBeenCalledOnce(); + expect(operations.posts.at(-1)?.text).toContain("New Pi session started"); + testRoute.actions.newSession = undefined; + await send("/new", "45.6"); + expect(operations.posts.at(-1)?.text).toContain("cannot start a new session remotely"); await send("/pause", "46"); expect(operations.posts.at(-1)?.text).toContain("paused"); await send("ordinary while paused", "47"); diff --git a/tests/state-store.test.ts b/tests/state-store.test.ts index dff0b7f..6824dd0 100644 --- a/tests/state-store.test.ts +++ b/tests/state-store.test.ts @@ -342,3 +342,49 @@ describe("TunnelStateStore", () => { expect(await store.getActiveChannelBindingForSession("slack", "session-1", { instanceId: "work", conversationId: "C1", userId: "U1" })).toBeUndefined(); }); }); + +describe("session binding migration", () => { + it("moves active bindings and selections while making old bindings stale", async () => { + const store = await createStore(); + await store.upsertBinding({ sessionKey: "old", sessionId: "old-id", sessionLabel: "old", chatId: 10, userId: 20, boundAt: "2026-01-01T00:00:00.000Z", lastSeenAt: "2026-01-01T00:00:00.000Z" }); + await store.upsertChannelBinding({ channel: "discord", instanceId: "default", conversationId: "C1", userId: "U1", sessionKey: "old", sessionId: "old-id", sessionLabel: "old", boundAt: "2026-01-01T00:00:00.000Z", lastSeenAt: "2026-01-01T00:00:00.000Z", paused: true }); + await store.setActiveChannelSelection("discord", "C1", "U1", "old"); + + const result = await store.migrateSessionBindings({ oldSessionKey: "old", newSessionKey: "new", newSessionId: "new-id", newSessionFile: "/sessions/new.jsonl", newSessionLabel: "new", now: "2026-01-02T00:00:00.000Z" }); + expect(result).toMatchObject({ telegram: [{ sessionKey: "new", status: "active" }], channels: [{ sessionKey: "new", status: "active", paused: true }], movedSelections: 2 }); + expect(await store.getBindingBySessionKey("old")).toMatchObject({ status: "revoked", movedToSessionKey: "new" }); + expect(await store.getChannelBindingRecordBySessionKey("discord", "old")).toMatchObject({ status: "revoked", movedToSessionKey: "new" }); + expect(await store.getActiveChannelSelection("telegram", "10", "20")).toMatchObject({ sessionKey: "new" }); + expect(await store.getActiveChannelSelection("discord", "C1", "U1")).toMatchObject({ sessionKey: "new" }); + }); + + it("does not overwrite revoked target records belonging to another identity", async () => { + const store = await createStore(); + await store.upsertBinding({ sessionKey: "old", sessionId: "old-id", sessionLabel: "old", chatId: 10, userId: 20, boundAt: "2026-01-01T00:00:00.000Z", lastSeenAt: "2026-01-01T00:00:00.000Z" }); + await store.upsertBinding({ sessionKey: "new", sessionId: "new-id", sessionLabel: "new", chatId: 99, userId: 98, boundAt: "2026-01-01T00:00:00.000Z", lastSeenAt: "2026-01-01T00:00:00.000Z" }); + await store.revokeBinding("new"); + await store.upsertChannelBinding({ channel: "slack", instanceId: "default", conversationId: "C-old", userId: "U-old", sessionKey: "old", sessionId: "old-id", sessionLabel: "old", boundAt: "2026-01-01T00:00:00.000Z", lastSeenAt: "2026-01-01T00:00:00.000Z" }); + await store.upsertChannelBinding({ channel: "slack", instanceId: "default", conversationId: "C-other", userId: "U-other", sessionKey: "new", sessionId: "new-id", sessionLabel: "new", boundAt: "2026-01-01T00:00:00.000Z", lastSeenAt: "2026-01-01T00:00:00.000Z" }); + await store.revokeChannelBinding("slack", "new"); + + const result = await store.migrateSessionBindings({ oldSessionKey: "old", newSessionKey: "new", newSessionId: "new-id", newSessionLabel: "new" }); + expect(result).toEqual({ telegram: [], channels: [], movedSelections: 0 }); + expect(await store.getBindingBySessionKey("new")).toMatchObject({ chatId: 99, userId: 98, status: "revoked" }); + expect(await store.getChannelBindingRecordBySessionKey("slack", "new")).toMatchObject({ conversationId: "C-other", userId: "U-other", status: "revoked" }); + }); + + it("fails closed for revoked bindings, conflicting target bindings, and requester mismatch", async () => { + const store = await createStore(); + await store.upsertBinding({ sessionKey: "old", sessionId: "old-id", sessionLabel: "old", chatId: 10, userId: 20, boundAt: "2026-01-01T00:00:00.000Z", lastSeenAt: "2026-01-01T00:00:00.000Z" }); + await store.upsertBinding({ sessionKey: "new", sessionId: "new-id", sessionLabel: "new", chatId: 99, userId: 99, boundAt: "2026-01-01T00:00:00.000Z", lastSeenAt: "2026-01-01T00:00:00.000Z" }); + await store.upsertChannelBinding({ channel: "slack", instanceId: "default", conversationId: "C1", userId: "U1", sessionKey: "old", sessionId: "old-id", sessionLabel: "old", boundAt: "2026-01-01T00:00:00.000Z", lastSeenAt: "2026-01-01T00:00:00.000Z" }); + + const result = await store.migrateSessionBindings({ oldSessionKey: "old", newSessionKey: "new", newSessionId: "new-id", newSessionLabel: "new", requester: { channel: "slack", conversationId: "other", userId: "U1" } }); + expect(result).toEqual({ telegram: [], channels: [], movedSelections: 0 }); + expect(await store.getBindingBySessionKey("old")).toMatchObject({ status: "active" }); + + await store.revokeChannelBinding("slack", "old"); + const revoked = await store.migrateSessionBindings({ oldSessionKey: "old", newSessionKey: "new-2", newSessionId: "new-2", newSessionLabel: "new-2", requester: { channel: "slack", conversationId: "C1", userId: "U1" } }); + expect(revoked.channels).toEqual([]); + }); +});