| /** | | | * DS Anchored Session Extensions | | | * | | | * Provides two commands that send a single "who are you" probe with: | | | * - a minimal system prompt ("You are a helpful software engineer assistant") | | | * - no tools in the provider payload | | | * | | | * /new-ds-anchored-session | | | * Creates a new session, then runs the anchored probe as its first message. | | | * | | | * /ds-anchor | | | * Runs the anchored probe in the current session. Fails if the session | | | * already has conversation history or the current model is not supported. | | | * | | | * The anchoring state is stored as a custom session entry so it survives | | | * session persistence; before_provider_request rewrites only the probe turn. | | | */ | | | import type { | | | ExtensionAPI, | | | ExtensionCommandContext, | | | ExtensionContext, | | | SessionEntry, | | | } from "@earendil-works/pi-coding-agent"; | | | const ALLOWED_PROVIDER = "deepseek"; | | | const ALLOWED_MODEL_IDS = new Set(["deepseek-v4-flash", "deepseek-v4-pro"]); | | | const SINGLE_SYSTEM_PROMPT = "You are a helpful software engineer assistant"; | |
| const ANCHOR_CUSTOM_TYPE = "ds-anchored-session"; | |
| const ANCHOR_WIDGET_ID = "ds-anchored-session"; | |
| interface AnchorState { | | | phase: "anchoring" | "done"; | | | } | | | interface ModelIdentity { | |
| provider: string; | |
| id: string; | |
| } | | | function isAllowedModel(model: ModelIdentity | undefined | null): boolean { | | | return ( | |
| model !== undefined && | |
| model !== null && | |
| model.provider === ALLOWED_PROVIDER && | |
| ALLOWED_MODEL_IDS.has(model.id) | |
| ); | |
| } | |
| function modelLabel(model: ModelIdentity): string { | |
| return `${model.provider}/${model.id}`; | |
| } | |
| function findLatestAnchorEntry(entries: SessionEntry[]): AnchorState | undefined { | |
| for (let i = entries.length - 1; i >= 0; i--) { | |
| const entry = entries[i]; | |
| if (entry.type === "custom" && entry.customType === ANCHOR_CUSTOM_TYPE) { | |
| return entry.data as AnchorState; | | | } | | | } | | | return undefined; | | | } | |
| function hasConversationHistory(entries: SessionEntry[]): boolean { | |
| return entries.some((entry) => entry.type === "message" || entry.type === "compaction"); | |
| } | |
| function isAnchoring(entries: SessionEntry[]): boolean { | |
| return findLatestAnchorEntry(entries)?.phase === "anchoring"; | |
| } | |
| function rewritePayloadForAnchoring(payload: unknown): unknown { | |
| if (typeof payload !== "object" || payload === null) { | |
| return payload; | | | } | | | const request = payload as Record<string, unknown>; | | | const messages = request.messages; | |
| if (Array.isArray(messages)) { | |
| const firstMessage = messages[0] as Record<string, unknown> | undefined; | |
| if ( | |
| firstMessage !== undefined && | |
| (firstMessage.role === "system" || firstMessage.role === "developer") | |
| ) { | |
| firstMessage.content = SINGLE_SYSTEM_PROMPT; | |
| } else { | |
| messages.unshift({ role: "system", content: SINGLE_SYSTEM_PROMPT }); | |
| } | | | } | | | // DeepSeek uses the OpenAI-compatible chat completions format. | | | delete request.tools; | | | delete request.tool_choice; | | | return request; | | | } | |
| export default function (pi: ExtensionAPI) { | |
| // Set by /new-ds-anchored-session before ctx.newSession() and consumed by | |
| // the replacement session's session_start handler. Plain module state is | | | // safe to carry across the session replacement lifecycle. | | | let pendingNewSessionAnchor = false; | | | pi.registerCommand("ds-anchor", { | | | description: | | | "Send a tool-free 'who are you' probe with a minimal system prompt in the current session", | |
| handler: async (_args: string, ctx: ExtensionCommandContext) => { | |
| if (ctx.model === undefined || !isAllowedModel(ctx.model)) { | |
| ctx.ui.notify( | |
| ds-anchor requires model ${ALLOWED_PROVIDER}/[${[...ALLOWED_MODEL_IDS].join(", ")}], current: ${ctx.model ? modelLabel(ctx.model) : "none"}, | |
| "error", | |
| ); | |
| return; | |
| } | |
| if (!ctx.modelRegistry.hasConfiguredAuth(ctx.model)) { | |
| ctx.ui.notify(`ds-anchor: no API key configured for ${modelLabel(ctx.model)}`, "error"); | |
| return; | | | } | | | if (hasConversationHistory(ctx.sessionManager.getEntries())) { | | | ctx.ui.notify("ds-anchor requires a session without conversation history", "error"); | | | return; | | | } | |
| pi.appendEntry(ANCHOR_CUSTOM_TYPE, { phase: "anchoring" }); | |
| ctx.ui.setWidget(ANCHOR_WIDGET_ID, ["Anchoring"]); | |
| pi.sendUserMessage("who are you"); | |
| }, | |
| }); | |
| pi.registerCommand("new-ds-anchored-session", { | |
| description: | | | "Create a new session and send a tool-free 'who are you' probe with a minimal system prompt", | |
| handler: async (_args: string, ctx: ExtensionCommandContext) => { | |
| pendingNewSessionAnchor = true; | |
| const result = await ctx.newSession({ | |
| withSession: async (newCtx) => { | |
| // session_start already wrote the anchoring state when the | |
| // replacement model is supported; this is the final check. | |
| if (newCtx.model === undefined || !isAllowedModel(newCtx.model)) { | |
| newCtx.ui.notify( | |
| new-ds-anchored-session requires model ${ALLOWED_PROVIDER}/[${[...ALLOWED_MODEL_IDS].join(", ")}], current: ${newCtx.model ? modelLabel(newCtx.model) : "none"}, | |
| "error", | |
| ); | |
| return; | |
| } | |
| if (!newCtx.modelRegistry.hasConfiguredAuth(newCtx.model)) { | |
| newCtx.ui.notify( | |
| new-ds-anchored-session: no API key configured for ${modelLabel(newCtx.model)}, | |
| "error", | |
| ); | |
| return; | |
| } | |
| await newCtx.sendUserMessage("who are you"); | |
| }, | |
| }); | |
| if (result.cancelled) { | |
| pendingNewSessionAnchor = false; | |
| } | | | }, | |
| }); | |
| pi.on("session_start", (event, ctx: ExtensionContext) => { | |
| if (event.reason !== "new" || !pendingNewSessionAnchor) { | |
| return; | | | } | | | pendingNewSessionAnchor = false; | | | // Only write the anchoring state when the replacement session is usable. | | | if (ctx.model === undefined || !isAllowedModel(ctx.model)) { | | | return; | | | } | | | if (!ctx.modelRegistry.hasConfiguredAuth(ctx.model)) { | | | return; | | | } | |
| pi.appendEntry(ANCHOR_CUSTOM_TYPE, { phase: "anchoring" }); | |
| ctx.ui.setWidget(ANCHOR_WIDGET_ID, ["Anchoring"]); | |
| }); | |
| pi.on("before_provider_request", (event, ctx: ExtensionContext) => { | |
| if (!isAnchoring(ctx.sessionManager.getEntries())) { | |
| return undefined; | | | } | | | // Safety net: never rewrite payloads for models outside the supported set. | | | if (!isAllowedModel(ctx.model)) { | | | return undefined; | | | } | |
| return rewritePayloadForAnchoring(event.payload); | |
| }); | |
| pi.on("agent_settled", (_event, ctx: ExtensionContext) => { | |
| if (!isAnchoring(ctx.sessionManager.getEntries())) { | |
| return; | | | } | |
| pi.appendEntry(ANCHOR_CUSTOM_TYPE, { phase: "done" }); | |
| ctx.ui.setWidget(ANCHOR_WIDGET_ID, ["Anchored"]); | |
| }); | |
| } |