feat(agy): queue turns, stream output, expose usage and image attachments - #1
Conversation
Code Review Completed! 🔥The code review was successfully completed based on your current configurations. Kody Guide: Usage and ConfigurationInteracting with Kody
Current Kody ConfigurationReview OptionsThe following review options are enabled or disabled:
|
github-actions
Bot
added
vouch:trusted
![kody-ai[bot]](https://avatars.githubusercontent.com/in/413034?s=60&v=4){kind=small}
| const deletePendingTurnStartByMessageId: ProjectionTurnRepositoryShape["deletePendingTurnStartByMessageId"] = | ||
| (input) => | ||
| deletePendingProjectionTurnByMessageId(input).pipe( | ||
| Effect.mapError( | ||
| toPersistenceSqlError( | ||
| "ProjectionTurnRepository.deletePendingTurnStartByMessageId:query", | ||
| ), | ||
| ), | ||
| ); |
There was a problem hiding this comment.
Stale pending-start placeholders can survive because deletePendingTurnStartByMessageId in apps/server/src/persistence/Layers/ProjectionTurns.ts has no production caller, so canceled or deleted queued prompts remain in projection_turns. getPendingTurnStartByThreadId then reuses the oldest pending row even after its message is gone, misattributing pendingMessageId and sourcePlans to a later turn; invoke deletePendingTurnStartByMessageId({ threadId, messageId }) on queued prompt deletion or filter for message liveness.
// Call deletePendingTurnStartByMessageId({ threadId, messageId }) whenever a
// queued user prompt is deleted/canceled so its pending placeholder cannot be
// consumed by a later turn; otherwise the FIFO getPendingTurnStartByThreadId
// keeps returning the stale row.Prompt for LLM
File apps/server/src/persistence/Layers/ProjectionTurns.ts:
Line 327 to 335:
Stale pending-start placeholders can survive because `deletePendingTurnStartByMessageId` in `apps/server/src/persistence/Layers/ProjectionTurns.ts` has no production caller, so canceled or deleted queued prompts remain in `projection_turns`. `getPendingTurnStartByThreadId` then reuses the oldest pending row even after its message is gone, misattributing `pendingMessageId` and `sourcePlans` to a later turn; invoke `deletePendingTurnStartByMessageId({ threadId, messageId })` on queued prompt deletion or filter for message liveness.
Suggested Code:
// Call deletePendingTurnStartByMessageId({ threadId, messageId }) whenever a
// queued user prompt is deleted/canceled so its pending placeholder cannot be
// consumed by a later turn; otherwise the FIFO getPendingTurnStartByThreadId
// keeps returning the stale row.
Talk to Kody by mentioning @kody
Was this suggestion helpful? React with 👍 or 👎 to help Kody learn from this interaction.
| if (stepType === "agent_response") { | ||
| const liveUsage = usageFromAgy(step.usage); | ||
| if (liveUsage !== undefined) { | ||
| yield* offerRuntimeEvent({ | ||
| type: "thread.token-usage.updated", | ||
| ...(yield* makeEventStamp()), | ||
| provider: PROVIDER, | ||
| providerInstanceId: boundInstanceId, | ||
| threadId: ctx.threadId, | ||
| turnId: turn.turnId, | ||
| payload: { usage: liveUsage }, | ||
| }); | ||
| } |
There was a problem hiding this comment.
Cumulative token accounting in the live agent_response path is incorrect because usageFromAgy(step.usage) ignores { cumulativeResult: true, previousCumulative: ctx.lastCumulativeUsage }, so mid-turn thread.token-usage.updated events overwrite lastInputTokens and lastCachedInputTokens with session-wide totals instead of per-turn deltas. Use usageFromAgy(step.usage, { cumulativeResult: true, previousCumulative: ctx.lastCumulativeUsage }) and refresh ctx.lastCumulativeUsage from agyUsageCounters(step.usage) when defined, matching the result handling.
const currentStepUsage = agyUsageCounters(step.usage);
const liveUsage = usageFromAgy(step.usage, {
cumulativeResult: true,
previousCumulative: ctx.lastCumulativeUsage,
});
if (currentStepUsage !== undefined) ctx.lastCumulativeUsage = currentStepUsage;Prompt for LLM
File apps/server/src/provider/Layers/AgyAdapter.ts:
Line 628 to 640:
Cumulative token accounting in the live `agent_response` path is incorrect because `usageFromAgy(step.usage)` ignores `{ cumulativeResult: true, previousCumulative: ctx.lastCumulativeUsage }`, so mid-turn `thread.token-usage.updated` events overwrite `lastInputTokens` and `lastCachedInputTokens` with session-wide totals instead of per-turn deltas. Use `usageFromAgy(step.usage, { cumulativeResult: true, previousCumulative: ctx.lastCumulativeUsage })` and refresh `ctx.lastCumulativeUsage` from `agyUsageCounters(step.usage)` when defined, matching the `result` handling.
Suggested Code:
const currentStepUsage = agyUsageCounters(step.usage);
const liveUsage = usageFromAgy(step.usage, {
cumulativeResult: true,
previousCumulative: ctx.lastCumulativeUsage,
});
if (currentStepUsage !== undefined) ctx.lastCumulativeUsage = currentStepUsage;
Talk to Kody by mentioning @kody
Was this suggestion helpful? React with 👍 or 👎 to help Kody learn from this interaction.
| if (status === "SUCCESS") { | ||
| yield* settleTurn(ctx, turn, { state: "completed", usage }); | ||
| yield* startNextQueuedTurn(ctx); | ||
| } else if (status === "CANCELED" || status === "INTERRUPTED") { |
There was a problem hiding this comment.
Queued turn loss in apps/server/src/provider/Layers/AgyAdapter.ts occurs when ctx.queuedTurns.length = 0 clears interrupted, cancelled, or failed prompts and clearOrConsumePendingProjectionTurnsByThread removes their pending rows without emitting any turn.completed or turn.failed event. Settle each dropped queued turn before clearing the queue, or emit a capped failure event per discarded prompt, so the UI surfaces an error instead of silently losing accepted input.
} else if (status === "CANCELED" || status === "INTERRUPTED") {
const dropped = [...ctx.queuedTurns];
ctx.queuedTurns.length = 0;
yield* Effect.forEach(dropped, (queued) =>
settleTurn(ctx, queued, { state: status === "CANCELED" ? "cancelled" : "interrupted" }),
);
yield* settleTurn(ctx, turn, {...});Prompt for LLM
File apps/server/src/provider/Layers/AgyAdapter.ts:
Line 601:
Queued turn loss in `apps/server/src/provider/Layers/AgyAdapter.ts` occurs when `ctx.queuedTurns.length = 0` clears interrupted, cancelled, or failed prompts and `clearOrConsumePendingProjectionTurnsByThread` removes their pending rows without emitting any `turn.completed` or `turn.failed` event. Settle each dropped queued turn before clearing the queue, or emit a capped failure event per discarded prompt, so the UI surfaces an error instead of silently losing accepted input.
Suggested Code:
} else if (status === "CANCELED" || status === "INTERRUPTED") {
const dropped = [...ctx.queuedTurns];
ctx.queuedTurns.length = 0;
yield* Effect.forEach(dropped, (queued) =>
settleTurn(ctx, queued, { state: status === "CANCELED" ? "cancelled" : "interrupted" }),
);
yield* settleTurn(ctx, turn, {...});
Talk to Kody by mentioning @kody
Was this suggestion helpful? React with 👍 or 👎 to help Kody learn from this interaction.
Resolved conflicts keeping both sides: fork Codex-session-import overlay (ws.ts, rpc.ts, server.test.ts, providers-codex.md) + upstream provider feedback upload (pingdotgg#7949) and app-approval docs (pingdotgg#8058). Fixed two latent PR #1 type errors surfaced by tsgo: - ContextWindowMeter.logic.ts: coalesce optional snapshot fields with ?? null - ProviderRuntimeIngestion.ts: drop Effect.fn.Return annotation pinning the error channel to never (ServerSettingsError now propagates like before)
1 participant
Overview
This PR enhances the Antigravity (agy) provider integration and the surrounding orchestration/UI with four related improvements: FIFO turn queueing, native streaming assistant output, granular token usage reporting, and image attachment support in headless text mode.
Key Changes
1. Queued turns: each prompt gets its own turn lifecycle
queuedTurns) so each user message becomes its own T3 turn, started only after the previous Antigravity turn emits its terminal result (SUCCESS/CANCELED/FAILED).projection_turnspending-start table was updated from a replace-style "latest row" model to an append-style FIFO queue:getPendingTurnStartByThreadIdnow returns the oldest pending start, anddeletePendingTurnStartByThreadIdconsumes only the oldest pending start while a session is running (clears all when the session is not running). A newdeletePendingTurnStartByMessageIdwas added for targeted cleanup.2. Streaming assistant output
assistantDeliveryMode: "streaming"in their capabilities. The Agy adapter advertises this capability.3. Token usage exposure
usageFromAgynow:cache_read_tokensback into canonical input/context token counts (matching how input context is commonly reported).agent_responsesteps and per-turn deltas on turn results.last*fields, which the UI now prefers over aggregate counters.agyas "Antigravity".4. Image attachments in headless text mode
agy --input-format stream-jsonaccepts text only, image attachments are resolved to validated local file paths and appended to the prompt as a delimited<t3_attached_images>manifest with metadata (name, MIME type, size, absolute path). Antigravity's native file/media tools can then inspect the referenced files.runtimeMode === "full-access"session now spawns the process with--dangerously-skip-permissions, enabling tool access in headless setups.Impact
The changes are localized to the Agy adapter, provider ingestion, projection turn persistence, and the context-window meter UI with matching unit tests.