mirror of
https://github.com/coder/coder.git
synced 2026-10-16 14:48:56 +08:00
feat(coderd): record provider response IDs on chat messages (#29763)
Chat messages could only be linked to AI Gateway records per chat, so nothing recorded which model response produced a given message. fantasy surfaces the response ID the model endpoint returns on the stream's finish part: the Anthropic Messages ID as the part ID, and the OpenAI Responses ID in provider metadata. chatd now stores that ID in the existing `chat_messages.provider_response_id` column for assistant steps and the compaction summary row. No migration and no aibridge changes. What the stored value is depends on the route: | Route | Stored value | Joins to | |---|---|---| | Anthropic Messages API through the AI Gateway (Anthropic and Bedrock Claude) | The gateway's interception ID. The gateway replaces the message ID because injected MCP tool loops can fold several upstream messages into one response. | `aibridge_interceptions.id` | | OpenAI Responses API through the AI Gateway | The upstream `resp_` ID, passed through unchanged | `provider_response_id` on `aibridge_token_usages`, `aibridge_user_prompts`, and `aibridge_tool_usages` | | Direct to the provider | The provider's own ID (`msg_`, `resp_`) | Nothing in the gateway tables | To resolve messages to interceptions, then read usage by `interception_id`: ```sql SELECT cm.id AS chat_message_id, i.id AS interception_id FROM chat_messages cm JOIN aibridge_interceptions i ON i.id::text = cm.provider_response_id UNION SELECT cm.id, tu.interception_id FROM chat_messages cm JOIN aibridge_token_usages tu ON tu.provider_response_id = cm.provider_response_id; ``` Not covered, and left NULL: - Partial rows saved on interrupt, because fantasy only reports the ID in the finish part. - Chat Completions providers (OpenAI-compatible, OpenRouter, Vercel), because fantasy does not surface the chunk ID there. - User, tool, and system rows, and all existing messages. <details> <summary>Delivery record</summary> - `96a896c` replaced the earlier interception-link design (`59fa58f`, `7f93c18`, `9d04ccd`) in place: the aibridge hook, the `aibridge_interception_id` column and migration, the message part buffer tracking, and the `ARCHITECTURE.md` section are all removed. The net diff against the base touches only `coderd/database` and `coderd/x/chatd`. - Local validation on `96a896c`: pre-commit hook green (gen, fmt, lint, build); `go test ./coderd/x/chatd/... ./coderd/database/` green. Red-green: removing the finish-part capture fails both new end-to-end tests (`TestActiveServer_PersistsAnthropicMessageID`, `TestOpenAIResponsesPersistsProviderResponseID`). CI on `96a896c`: green. - Remote UAT round 2 on `96a896c` ([chat](https://dogfood.cdr.dev/agents/75353a23-cd9e-4c81-a99f-472c78f3ba32)): **FAIL** against this description's earlier claim that every stored ID joins the gateway usage rows directly. Anthropic rows routed through the gateway stored the interception ID, so 0 of 18 joined usage rows directly, while every checked Anthropic row equaled `aibridge_interceptions.id`. Both OpenAI Responses rows joined directly. Interrupted partial rows, Chat Completions rows, and user and tool rows were NULL, with no empty strings or duplicate IDs. Multi-step turns, manual compaction, edit and regenerate, a provider switch, and a retried 429 behaved as expected. One unattributed anomaly: a single duplicate user row after interrupt and continue, which did not reproduce and is outside this diff. - `2e889ba` corrects the doc comment on `PersistedStep.ProviderResponseID` and this description to match; behavior is unchanged. Pre-commit hook and CI green. - Remote UAT round 3 on `2e889ba` against the corrected contract ([same chat](https://dogfood.cdr.dev/agents/75353a23-cd9e-4c81-a99f-472c78f3ba32)): **PASS**. The query above resolved all 40 stamped rows to exactly one interception from the same chat, with none unresolved. Reading usage by interception recovered every usage row for 38 of them; the last two resolved but their usage rows were not counted. Fresh Anthropic and Responses calls, a seven-step tool turn, manual compaction, an Anthropic to Responses to Anthropic switch in one chat, and a UI reload behaved as described, and the NULL cases held. Three interrupt and continue attempts did not reproduce the round-2 duplicate, so it stays unattributed. Not tested: gateway-injected MCP tool loops (unreachable from chatd, whose delegated requests skip the gateway's MCP proxy), direct Anthropic `msg_` IDs, Bedrock, and threshold-triggered compaction. Both rounds had a verified isolation proof and a clean dogfood audit. - Codex rounds 1 to 3 reviewed the old design. Codex round 4 on `2e889ba` ([request](https://github.com/coder/coder/pull/29763#issuecomment-5796193371), 13:58Z): no major issues ([result](https://github.com/coder/coder/pull/29763#issuecomment-5796267231), reviewed commit `2e889ba05c`). No unresolved review threads. Review passes so far: 4 of 6. - Final head `2e889ba` is the exact commit remote UAT round 3 and Codex round 4 covered; required checks green. </details> > Xum updated this PR on behalf of @ibetitsmike.
This commit is contained in:
@@ -141,6 +141,7 @@ func ChatMessage(t testing.TB, db database.Store, seed database.ChatMessage) dat
|
||||
ContextLimit: []int64{seed.ContextLimit.Int64},
|
||||
Compressed: []bool{seed.Compressed},
|
||||
RuntimeMs: []int64{seed.RuntimeMs.Int64},
|
||||
ProviderResponseID: []string{seed.ProviderResponseID.String},
|
||||
})
|
||||
require.NoError(t, err, "insert chat message")
|
||||
require.Len(t, msgs, 1)
|
||||
|
||||
Generated
+6
-2
@@ -11096,7 +11096,8 @@ inserted AS (
|
||||
cache_read_tokens,
|
||||
context_limit,
|
||||
compressed,
|
||||
runtime_ms
|
||||
runtime_ms,
|
||||
provider_response_id
|
||||
)
|
||||
SELECT
|
||||
allocated.id,
|
||||
@@ -11116,7 +11117,8 @@ inserted AS (
|
||||
NULLIF(($14::bigint[])[allocated.ord], 0),
|
||||
NULLIF(($15::bigint[])[allocated.ord], 0),
|
||||
($16::boolean[])[allocated.ord],
|
||||
NULLIF(($17::bigint[])[allocated.ord], 0)
|
||||
NULLIF(($17::bigint[])[allocated.ord], 0),
|
||||
NULLIF(($18::text[])[allocated.ord], '')
|
||||
FROM allocated
|
||||
RETURNING id, chat_id, model_config_id, created_at, role, content, visibility, input_tokens, output_tokens, total_tokens, reasoning_tokens, cache_creation_tokens, cache_read_tokens, context_limit, compressed, created_by, content_version, total_cost_micros, runtime_ms, deleted, provider_response_id, revision, reasoning_effort, search_tsv, search_tsv_config
|
||||
)
|
||||
@@ -11143,6 +11145,7 @@ type InsertChatMessagesParams struct {
|
||||
ContextLimit []int64 `db:"context_limit" json:"context_limit"`
|
||||
Compressed []bool `db:"compressed" json:"compressed"`
|
||||
RuntimeMs []int64 `db:"runtime_ms" json:"runtime_ms"`
|
||||
ProviderResponseID []string `db:"provider_response_id" json:"provider_response_id"`
|
||||
}
|
||||
|
||||
type InsertChatMessagesRow struct {
|
||||
@@ -11195,6 +11198,7 @@ func (q *sqlQuerier) InsertChatMessages(ctx context.Context, arg InsertChatMessa
|
||||
pq.Array(arg.ContextLimit),
|
||||
pq.Array(arg.Compressed),
|
||||
pq.Array(arg.RuntimeMs),
|
||||
pq.Array(arg.ProviderResponseID),
|
||||
)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
|
||||
@@ -954,7 +954,8 @@ inserted AS (
|
||||
cache_read_tokens,
|
||||
context_limit,
|
||||
compressed,
|
||||
runtime_ms
|
||||
runtime_ms,
|
||||
provider_response_id
|
||||
)
|
||||
SELECT
|
||||
allocated.id,
|
||||
@@ -974,7 +975,8 @@ inserted AS (
|
||||
NULLIF((@cache_read_tokens::bigint[])[allocated.ord], 0),
|
||||
NULLIF((@context_limit::bigint[])[allocated.ord], 0),
|
||||
(@compressed::boolean[])[allocated.ord],
|
||||
NULLIF((@runtime_ms::bigint[])[allocated.ord], 0)
|
||||
NULLIF((@runtime_ms::bigint[])[allocated.ord], 0),
|
||||
NULLIF((@provider_response_id::text[])[allocated.ord], '')
|
||||
FROM allocated
|
||||
RETURNING *
|
||||
)
|
||||
|
||||
@@ -20,10 +20,11 @@ const (
|
||||
|
||||
// stepData is the durable content produced by one provider attempt.
|
||||
type stepData struct {
|
||||
Content []fantasy.Content
|
||||
Usage fantasy.Usage
|
||||
ContextLimit sql.NullInt64
|
||||
Runtime time.Duration
|
||||
Content []fantasy.Content
|
||||
Usage fantasy.Usage
|
||||
ContextLimit sql.NullInt64
|
||||
Runtime time.Duration
|
||||
ProviderResponseID string
|
||||
|
||||
// BatchRuntime is the local-tool batch window. Model steps use Runtime.
|
||||
BatchRuntime time.Duration
|
||||
@@ -57,6 +58,7 @@ type compactionOutcome struct {
|
||||
ContextLimit int64
|
||||
EstimatedContextTokens int64
|
||||
Runtime time.Duration
|
||||
ProviderResponseID string
|
||||
}
|
||||
|
||||
type compactionStatus int
|
||||
|
||||
@@ -5263,6 +5263,37 @@ func readFileResponseWithInputTokens(path string, inputTokens int) chattest.Anth
|
||||
return chattest.AnthropicStreamingResponse(chunks...)
|
||||
}
|
||||
|
||||
func TestActiveServer_PersistsAnthropicMessageID(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
ctx := testutil.Context(t, testutil.WaitLong)
|
||||
db, ps := dbtestutil.NewDB(t)
|
||||
anthropicURL := chattest.NewAnthropic(t, func(req *chattest.AnthropicRequest) chattest.AnthropicResponse {
|
||||
if !req.Stream {
|
||||
return chattest.AnthropicNonStreamingResponse("title")
|
||||
}
|
||||
chunks := chattest.AnthropicTextChunks("answer")
|
||||
chunks[0].Message.ID = "msg_persisted"
|
||||
return chattest.AnthropicStreamingResponse(chunks...)
|
||||
})
|
||||
user, org, model := seedAnthropicChatDependencies(t, db, anthropicURL)
|
||||
server := newActiveTestServer(t, db, ps, func(cfg *chatd.Config) {
|
||||
cfg.AIBridgeTransportFactory = chatAIGatewayTransportFactoryPointer(chattest.NewMockAIBridgeTransport(t, anthropicURL, chattest.WithPreservePath()))
|
||||
})
|
||||
|
||||
chat := createChatThroughServer(ctx, t, db, server, org.ID, user.ID, model.ID, "hello")
|
||||
waitForChatStatus(ctx, t, db, chat.ID, database.ChatStatusWaiting)
|
||||
|
||||
messages, err := db.GetChatMessagesByChatID(ctx, database.GetChatMessagesByChatIDParams{ChatID: chat.ID})
|
||||
require.NoError(t, err)
|
||||
responseIDs := make(map[database.ChatMessageRole]sql.NullString)
|
||||
for _, message := range messages {
|
||||
responseIDs[message.Role] = message.ProviderResponseID
|
||||
}
|
||||
require.Equal(t, sql.NullString{String: "msg_persisted", Valid: true}, responseIDs[database.ChatMessageRoleAssistant])
|
||||
require.False(t, responseIDs[database.ChatMessageRoleUser].Valid)
|
||||
}
|
||||
|
||||
func TestActiveServer_RoutingPreservesAPIKeyAfterCompaction(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
|
||||
@@ -17,6 +17,7 @@ import (
|
||||
|
||||
"charm.land/fantasy"
|
||||
fantasyanthropic "charm.land/fantasy/providers/anthropic"
|
||||
fantasyopenai "charm.land/fantasy/providers/openai"
|
||||
"github.com/google/uuid"
|
||||
"golang.org/x/xerrors"
|
||||
|
||||
@@ -66,6 +67,10 @@ type PersistedStep struct {
|
||||
// Runtime is the wall-clock duration from opening to consuming the
|
||||
// model stream.
|
||||
Runtime time.Duration
|
||||
// ProviderResponseID is the response ID the model endpoint reported, if
|
||||
// any. Through the AI Gateway, Anthropic Messages responses carry the
|
||||
// gateway's interception ID instead of the upstream message ID.
|
||||
ProviderResponseID string
|
||||
// BatchRuntime is the union of billed local-tool execution intervals.
|
||||
// Parallel calls count once and serial calls count from their own start.
|
||||
BatchRuntime time.Duration
|
||||
@@ -263,6 +268,7 @@ type stepResult struct {
|
||||
content []fantasy.Content
|
||||
usage fantasy.Usage
|
||||
providerMetadata fantasy.ProviderMetadata
|
||||
providerResponseID string
|
||||
finishReason fantasy.FinishReason
|
||||
toolCalls []fantasy.ToolCallContent
|
||||
toolCallCreatedAt map[string]time.Time
|
||||
@@ -380,6 +386,7 @@ func GenerateAssistant(ctx context.Context, opts GenerateAssistantOptions) (Assi
|
||||
Usage: result.usage,
|
||||
ContextLimit: contextLimit,
|
||||
Runtime: opts.Clock.Since(stepStart),
|
||||
ProviderResponseID: result.providerResponseID,
|
||||
ToolCallCreatedAt: result.toolCallCreatedAt,
|
||||
ToolResultCreatedAt: result.toolResultCreatedAt,
|
||||
ReasoningStartedAt: result.reasoningStartedAt,
|
||||
@@ -999,6 +1006,7 @@ func processStepStream(
|
||||
result.usage = part.Usage
|
||||
result.finishReason = part.FinishReason
|
||||
result.providerMetadata = part.ProviderMetadata
|
||||
result.providerResponseID = providerResponseID(part)
|
||||
|
||||
case fantasy.StreamPartTypeError:
|
||||
return result, part.Error
|
||||
@@ -1008,6 +1016,22 @@ func processStepStream(
|
||||
return result, nil
|
||||
}
|
||||
|
||||
// providerResponseID returns the response ID from a finish part.
|
||||
// Anthropic and Bedrock report the message ID as the part ID, and the OpenAI
|
||||
// Responses API reports it in provider metadata. Chat Completions providers
|
||||
// report neither.
|
||||
func providerResponseID(part fantasy.StreamPart) string {
|
||||
if part.ID != "" {
|
||||
return part.ID
|
||||
}
|
||||
for _, metadata := range part.ProviderMetadata {
|
||||
if responses, ok := metadata.(*fantasyopenai.ResponsesProviderMetadata); ok && responses != nil {
|
||||
return responses.ResponseID
|
||||
}
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
type toolExecutionResult struct {
|
||||
content fantasy.ToolResultContent
|
||||
interval BilledInterval
|
||||
|
||||
@@ -18,6 +18,7 @@ import (
|
||||
|
||||
"charm.land/fantasy"
|
||||
fantasyanthropic "charm.land/fantasy/providers/anthropic"
|
||||
fantasyopenai "charm.land/fantasy/providers/openai"
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
promtestutil "github.com/prometheus/client_golang/prometheus/testutil"
|
||||
"github.com/stretchr/testify/assert"
|
||||
@@ -176,6 +177,60 @@ func TestGenerateAssistant_ProviderContextSurvivesStreamError(t *testing.T) {
|
||||
require.Equal(t, "OpenAI returned an unexpected error.", classified.Message)
|
||||
}
|
||||
|
||||
func TestGenerateAssistant_ProviderResponseID(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
cases := []struct {
|
||||
name string
|
||||
finish fantasy.StreamPart
|
||||
want string
|
||||
}{
|
||||
{
|
||||
name: "anthropic message ID",
|
||||
finish: fantasy.StreamPart{ID: "msg_anthropic"},
|
||||
want: "msg_anthropic",
|
||||
},
|
||||
{
|
||||
name: "openai responses metadata",
|
||||
finish: fantasy.StreamPart{ProviderMetadata: fantasy.ProviderMetadata{
|
||||
fantasyopenai.Name: &fantasyopenai.ResponsesProviderMetadata{ResponseID: "resp_openai"},
|
||||
}},
|
||||
want: "resp_openai",
|
||||
},
|
||||
{
|
||||
name: "not reported",
|
||||
},
|
||||
}
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
finish := tc.finish
|
||||
finish.Type = fantasy.StreamPartTypeFinish
|
||||
finish.FinishReason = fantasy.FinishReasonStop
|
||||
model := &chattest.FakeModel{
|
||||
ProviderName: "fake",
|
||||
ModelName: "fake-model",
|
||||
StreamFn: func(_ context.Context, _ fantasy.Call) (fantasy.StreamResponse, error) {
|
||||
return streamFromParts([]fantasy.StreamPart{
|
||||
{Type: fantasy.StreamPartTypeTextStart, ID: "text"},
|
||||
{Type: fantasy.StreamPartTypeTextDelta, ID: "text", Delta: "hi"},
|
||||
{Type: fantasy.StreamPartTypeTextEnd, ID: "text"},
|
||||
finish,
|
||||
}), nil
|
||||
},
|
||||
}
|
||||
|
||||
outcome, err := GenerateAssistant(context.Background(), GenerateAssistantOptions{
|
||||
Model: model,
|
||||
Messages: []fantasy.Message{textMessage(fantasy.MessageRoleUser, "hello")},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, tc.want, outcome.Step.ProviderResponseID)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestGenerateAssistant_ErrorProviderOverridesTransportLabel(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
|
||||
@@ -143,6 +143,9 @@ type CompactionResult struct {
|
||||
// PersistedStep.Runtime). Zero when the run was gated off before
|
||||
// calling the model.
|
||||
Runtime time.Duration
|
||||
// ProviderResponseID identifies the summary response. See
|
||||
// PersistedStep.ProviderResponseID.
|
||||
ProviderResponseID string
|
||||
}
|
||||
|
||||
// GenerateCompaction generates one context summary and returns it without
|
||||
@@ -207,7 +210,7 @@ func GenerateCompaction(ctx context.Context, opts GenerateCompactionOptions) (Co
|
||||
if opts.OnModelStreamStart != nil {
|
||||
opts.OnModelStreamStart()
|
||||
}
|
||||
summary, err := generateCompactionSummary(ctx, opts.Model, opts.Messages, config)
|
||||
summary, responseID, err := generateCompactionSummary(ctx, opts.Model, opts.Messages, config)
|
||||
if err != nil {
|
||||
publishCompactionError(config, "failed to generate compaction summary")
|
||||
return CompactionResult{}, err
|
||||
@@ -222,13 +225,14 @@ func GenerateCompaction(ctx context.Context, opts GenerateCompactionOptions) (Co
|
||||
SystemSummary: strings.TrimSpace(
|
||||
config.SystemSummaryPrefix + "\n\n" + summary,
|
||||
),
|
||||
SummaryReport: summary,
|
||||
Source: config.Source,
|
||||
ThresholdPercent: config.ThresholdPercent,
|
||||
UsagePercent: usagePercent,
|
||||
ContextTokens: contextTokens,
|
||||
ContextLimit: contextLimit,
|
||||
Runtime: summaryRuntime,
|
||||
SummaryReport: summary,
|
||||
Source: config.Source,
|
||||
ThresholdPercent: config.ThresholdPercent,
|
||||
UsagePercent: usagePercent,
|
||||
ContextTokens: contextTokens,
|
||||
ContextLimit: contextLimit,
|
||||
Runtime: summaryRuntime,
|
||||
ProviderResponseID: responseID,
|
||||
}
|
||||
result.EstimatedContextTokens = int64((len(result.SystemSummary) + bytesPerTokenEstimate - 1) / bytesPerTokenEstimate)
|
||||
if config.PublishMessagePart != nil && config.ToolCallID != "" {
|
||||
@@ -481,7 +485,7 @@ func generateCompactionSummary(
|
||||
model fantasy.LanguageModel,
|
||||
messages []fantasy.Message,
|
||||
options CompactionOptions,
|
||||
) (summary string, err error) {
|
||||
) (summary string, responseID string, err error) {
|
||||
summaryPrompt := make([]fantasy.Message, 0, len(messages)+1)
|
||||
summaryPrompt = append(summaryPrompt, messages...)
|
||||
summaryParts := []fantasy.MessagePart{fantasy.TextPart{Text: options.SummaryPrompt}}
|
||||
@@ -575,6 +579,7 @@ func generateCompactionSummary(
|
||||
case fantasy.StreamPartTypeFinish:
|
||||
finishSeen = true
|
||||
finishReason = part.FinishReason
|
||||
responseID = providerResponseID(part)
|
||||
case fantasy.StreamPartTypeError:
|
||||
streamErr = part.Error
|
||||
if streamErr == nil {
|
||||
@@ -626,7 +631,7 @@ func generateCompactionSummary(
|
||||
call.Tools = nil
|
||||
summary, err = streamSummaryText()
|
||||
}
|
||||
return summary, err
|
||||
return summary, responseID, err
|
||||
}
|
||||
|
||||
// contextTooLargePhrases are context-window rejections fantasy does not
|
||||
|
||||
@@ -42,7 +42,7 @@ func summaryTextStream() fantasy.StreamResponse {
|
||||
fantasy.StreamPart{Type: fantasy.StreamPartTypeTextStart, ID: "text"},
|
||||
fantasy.StreamPart{Type: fantasy.StreamPartTypeTextDelta, ID: "text", Delta: "summary"},
|
||||
fantasy.StreamPart{Type: fantasy.StreamPartTypeTextEnd, ID: "text"},
|
||||
fantasy.StreamPart{Type: fantasy.StreamPartTypeFinish, FinishReason: fantasy.FinishReasonStop},
|
||||
fantasy.StreamPart{Type: fantasy.StreamPartTypeFinish, ID: "msg_summary", FinishReason: fantasy.FinishReasonStop},
|
||||
)
|
||||
}
|
||||
|
||||
@@ -221,7 +221,7 @@ func TestGenerateCompactionSummary_PanicFinalizesAsError(t *testing.T) {
|
||||
})
|
||||
|
||||
require.PanicsWithValue(t, "compaction model crash", func() {
|
||||
_, _ = generateCompactionSummary(parentCtx, model,
|
||||
_, _, _ = generateCompactionSummary(parentCtx, model,
|
||||
[]fantasy.Message{textMessage(fantasy.MessageRoleUser, "hello")},
|
||||
CompactionOptions{
|
||||
DebugSvc: svc,
|
||||
@@ -273,7 +273,7 @@ func TestGenerateCompactionSummaryPreservesCallOptions(t *testing.T) {
|
||||
}
|
||||
|
||||
toolChoice := fantasy.ToolChoiceNone
|
||||
summary, err := generateCompactionSummary(context.Background(), model, messages, CompactionOptions{
|
||||
summary, _, err := generateCompactionSummary(context.Background(), model, messages, CompactionOptions{
|
||||
SummaryPrompt: "summarize",
|
||||
SummaryCall: fantasy.Call{
|
||||
Temperature: &temperature,
|
||||
@@ -329,7 +329,7 @@ func TestGenerateCompactionSummaryUsesToolDefinitions(t *testing.T) {
|
||||
}
|
||||
|
||||
toolChoice := fantasy.ToolChoiceNone
|
||||
summary, err := generateCompactionSummary(context.Background(), model, messages, CompactionOptions{
|
||||
summary, _, err := generateCompactionSummary(context.Background(), model, messages, CompactionOptions{
|
||||
SummaryPrompt: "summarize",
|
||||
SummaryCall: fantasy.Call{ToolChoice: &toolChoice},
|
||||
ToolDefinitions: toolDefinitions,
|
||||
@@ -362,7 +362,7 @@ func TestGenerateCompactionSummaryAppliesAnthropicPromptCaching(t *testing.T) {
|
||||
},
|
||||
}
|
||||
|
||||
_, err := generateCompactionSummary(context.Background(), model, messages, CompactionOptions{
|
||||
_, _, err := generateCompactionSummary(context.Background(), model, messages, CompactionOptions{
|
||||
SummaryPrompt: "summarize",
|
||||
})
|
||||
require.NoError(t, err)
|
||||
@@ -404,7 +404,7 @@ func TestGenerateCompactionSummaryRetriesWithoutToolsWhenContextTooLarge(t *test
|
||||
},
|
||||
}
|
||||
|
||||
summary, err := generateCompactionSummary(context.Background(), model,
|
||||
summary, responseID, err := generateCompactionSummary(context.Background(), model,
|
||||
[]fantasy.Message{textMessage(fantasy.MessageRoleUser, "hello")},
|
||||
CompactionOptions{
|
||||
SummaryPrompt: "summarize",
|
||||
@@ -414,6 +414,7 @@ func TestGenerateCompactionSummaryRetriesWithoutToolsWhenContextTooLarge(t *test
|
||||
)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, "summary", summary)
|
||||
require.Equal(t, "msg_summary", responseID)
|
||||
require.Len(t, calls, 2)
|
||||
require.Equal(t, toolDefinitions, calls[0].Tools)
|
||||
require.Nil(t, calls[1].Tools)
|
||||
@@ -473,7 +474,7 @@ func TestGenerateCompactionSummaryDoesNotRetryOtherFailures(t *testing.T) {
|
||||
},
|
||||
}
|
||||
|
||||
_, err := generateCompactionSummary(context.Background(), model,
|
||||
_, _, err := generateCompactionSummary(context.Background(), model,
|
||||
[]fantasy.Message{textMessage(fantasy.MessageRoleUser, "hello")},
|
||||
CompactionOptions{
|
||||
SummaryPrompt: "summarize",
|
||||
@@ -603,7 +604,7 @@ func TestGenerateCompactionSummary_UsesCallerContext(t *testing.T) {
|
||||
},
|
||||
}
|
||||
|
||||
summary, err := generateCompactionSummary(testCtx, model,
|
||||
summary, _, err := generateCompactionSummary(testCtx, model,
|
||||
[]fantasy.Message{textMessage(fantasy.MessageRoleUser, "hello")},
|
||||
CompactionOptions{SummaryPrompt: "summarize"},
|
||||
)
|
||||
@@ -641,7 +642,7 @@ func TestGenerateCompactionSummary_Stream(t *testing.T) {
|
||||
},
|
||||
}
|
||||
|
||||
summary, err := generateCompactionSummary(context.Background(), model, nil, CompactionOptions{})
|
||||
summary, _, err := generateCompactionSummary(context.Background(), model, nil, CompactionOptions{})
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, "first second", summary)
|
||||
})
|
||||
@@ -661,7 +662,7 @@ func TestGenerateCompactionSummary_Stream(t *testing.T) {
|
||||
},
|
||||
}
|
||||
|
||||
summary, err := generateCompactionSummary(context.Background(), model, nil, CompactionOptions{})
|
||||
summary, _, err := generateCompactionSummary(context.Background(), model, nil, CompactionOptions{})
|
||||
require.Empty(t, summary)
|
||||
require.EqualError(t, err, "compaction summary was truncated at the output token cap")
|
||||
})
|
||||
@@ -679,7 +680,7 @@ func TestGenerateCompactionSummary_Stream(t *testing.T) {
|
||||
},
|
||||
}
|
||||
|
||||
summary, err := generateCompactionSummary(context.Background(), model, nil, CompactionOptions{})
|
||||
summary, _, err := generateCompactionSummary(context.Background(), model, nil, CompactionOptions{})
|
||||
require.Empty(t, summary)
|
||||
require.EqualError(t, err, "compaction summary stream ended without a finish part")
|
||||
})
|
||||
@@ -694,7 +695,7 @@ func TestGenerateCompactionSummary_Stream(t *testing.T) {
|
||||
},
|
||||
}
|
||||
|
||||
summary, err := generateCompactionSummary(context.Background(), model, nil, CompactionOptions{
|
||||
summary, _, err := generateCompactionSummary(context.Background(), model, nil, CompactionOptions{
|
||||
ResolvedProvider: "fake",
|
||||
})
|
||||
require.Empty(t, summary)
|
||||
@@ -716,7 +717,7 @@ func TestGenerateCompactionSummary_Stream(t *testing.T) {
|
||||
},
|
||||
}
|
||||
|
||||
summary, err := generateCompactionSummary(context.Background(), model, nil, CompactionOptions{
|
||||
summary, _, err := generateCompactionSummary(context.Background(), model, nil, CompactionOptions{
|
||||
ResolvedProvider: "fake",
|
||||
})
|
||||
require.Empty(t, summary)
|
||||
@@ -744,7 +745,7 @@ func TestGenerateCompactionSummary_Stream(t *testing.T) {
|
||||
|
||||
done := make(chan error, 1)
|
||||
go func() {
|
||||
_, err := generateCompactionSummary(context.Background(), model, nil, CompactionOptions{
|
||||
_, _, err := generateCompactionSummary(context.Background(), model, nil, CompactionOptions{
|
||||
ResolvedProvider: "openai",
|
||||
Clock: clock,
|
||||
StreamSilenceTimeout: 5 * time.Millisecond,
|
||||
|
||||
@@ -34,6 +34,7 @@ type Message struct {
|
||||
CacheReadTokens sql.NullInt64
|
||||
ContextLimit sql.NullInt64
|
||||
RuntimeMs sql.NullInt64
|
||||
ProviderResponseID sql.NullString
|
||||
}
|
||||
|
||||
// toInsertParams converts a batch of Messages into the parallel-array
|
||||
@@ -62,6 +63,7 @@ func toInsertParams(chatID uuid.UUID, messages []Message) database.InsertChatMes
|
||||
ContextLimit: make([]int64, n),
|
||||
Compressed: make([]bool, n),
|
||||
RuntimeMs: make([]int64, n),
|
||||
ProviderResponseID: make([]string, n),
|
||||
}
|
||||
for i, m := range messages {
|
||||
params.CreatedBy[i] = nullUUIDOrNil(m.CreatedBy)
|
||||
@@ -88,6 +90,7 @@ func toInsertParams(chatID uuid.UUID, messages []Message) database.InsertChatMes
|
||||
params.ContextLimit[i] = nullInt64Or(m.ContextLimit, 0)
|
||||
params.Compressed[i] = m.Compressed
|
||||
params.RuntimeMs[i] = nullInt64Or(m.RuntimeMs, 0)
|
||||
params.ProviderResponseID[i] = m.ProviderResponseID.String
|
||||
}
|
||||
return params
|
||||
}
|
||||
|
||||
@@ -1592,6 +1592,7 @@ func stepDataFromPersisted(step chatloop.PersistedStep) stepData {
|
||||
Usage: step.Usage,
|
||||
ContextLimit: step.ContextLimit,
|
||||
Runtime: step.Runtime,
|
||||
ProviderResponseID: step.ProviderResponseID,
|
||||
BatchRuntime: step.BatchRuntime,
|
||||
BatchBilledCalls: step.BatchBilledCalls,
|
||||
ToolCallCreatedAt: step.ToolCallCreatedAt,
|
||||
|
||||
@@ -2,6 +2,7 @@ package chatd_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"fmt"
|
||||
"strings"
|
||||
"sync"
|
||||
@@ -105,6 +106,51 @@ func TestOpenAIResponsesNoStaleWebSearchReplay(t *testing.T) {
|
||||
require.NotContains(t, promptItemTypes(followup.Prompt), "web_search_call")
|
||||
}
|
||||
|
||||
func TestOpenAIResponsesPersistsProviderResponseID(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
db, ps := dbtestutil.NewDB(t)
|
||||
ctx := testutil.Context(t, testutil.WaitLong)
|
||||
|
||||
openAIURL := chattest.NewOpenAI(t, func(req *chattest.OpenAIRequest) chattest.OpenAIResponse {
|
||||
if !req.Stream {
|
||||
return chattest.OpenAINonStreamingResponse("title")
|
||||
}
|
||||
resp := chattest.OpenAIStreamingResponse(chattest.OpenAITextChunks("answer")...)
|
||||
resp.ResponseID = "resp_persisted"
|
||||
return resp
|
||||
})
|
||||
|
||||
user, org, _ := seedChatDependenciesWithProvider(t, db, "openai", openAIURL)
|
||||
model := insertOpenAIResponsesModelConfig(t, db, user.ID, false, false)
|
||||
factory := chattest.NewMockAIBridgeTransport(t, openAIURL)
|
||||
server := newOpenAIResponsesTestServer(t, db, ps, func(cfg *chatd.Config) {
|
||||
cfg.AIBridgeTransportFactory = chatAIGatewayTransportFactoryPointer(factory)
|
||||
})
|
||||
|
||||
chat, err := server.CreateChat(ctx, chatd.CreateOptions{
|
||||
OrganizationID: org.ID,
|
||||
OwnerID: user.ID,
|
||||
Title: uniqueResponsesTitle(t, "response-id"),
|
||||
ModelConfigID: model.ID,
|
||||
InitialUserContent: []codersdk.ChatMessagePart{
|
||||
codersdk.ChatMessageText("hello"),
|
||||
},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
waitForChatProcessed(ctx, t, db, chat.ID, server)
|
||||
requireResponsesChatWaiting(ctx, t, db, chat.ID)
|
||||
|
||||
messages, err := db.GetChatMessagesByChatID(ctx, database.GetChatMessagesByChatIDParams{ChatID: chat.ID})
|
||||
require.NoError(t, err)
|
||||
responseIDs := make(map[database.ChatMessageRole]sql.NullString)
|
||||
for _, message := range messages {
|
||||
responseIDs[message.Role] = message.ProviderResponseID
|
||||
}
|
||||
require.Equal(t, sql.NullString{String: "resp_persisted", Valid: true}, responseIDs[database.ChatMessageRoleAssistant])
|
||||
require.False(t, responseIDs[database.ChatMessageRoleUser].Valid)
|
||||
}
|
||||
|
||||
func TestOpenAIResponsesFullReplayPairsReasoningAndWebSearch(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
|
||||
@@ -207,6 +207,7 @@ func assistantMessage(
|
||||
// invocation shorter than a millisecond persists the same way an
|
||||
// unmeasured one does.
|
||||
msg.RuntimeMs = nullInt64IfNonZero(step.Runtime.Milliseconds())
|
||||
msg.ProviderResponseID = sql.NullString{String: step.ProviderResponseID, Valid: step.ProviderResponseID != ""}
|
||||
return msg
|
||||
}
|
||||
|
||||
@@ -356,6 +357,7 @@ func buildCompactionMessages(input buildCompactionMessagesInput) (compactionMess
|
||||
|
||||
assistantMsg := baseMessage(database.ChatMessageRoleAssistant, database.ChatMessageVisibilityUser, input.modelConfigID, contentVersion, assistantContent)
|
||||
assistantMsg.RuntimeMs = nullInt64IfNonZero(input.compaction.Runtime.Milliseconds())
|
||||
assistantMsg.ProviderResponseID = sql.NullString{String: input.compaction.ProviderResponseID, Valid: input.compaction.ProviderResponseID != ""}
|
||||
messages := []chatstate.Message{
|
||||
{
|
||||
Role: database.ChatMessageRoleUser,
|
||||
|
||||
@@ -239,10 +239,11 @@ func TestBuildCommitStepMessages_UsageRuntime(t *testing.T) {
|
||||
contentVersion: chatprompt.CurrentContentVersion,
|
||||
logger: slog.Make(),
|
||||
step: stepData{
|
||||
Content: []fantasy.Content{fantasy.TextContent{Text: "usage"}},
|
||||
Usage: fantasy.Usage{InputTokens: 100, OutputTokens: 20, TotalTokens: 120, ReasoningTokens: 3, CacheCreationTokens: 4, CacheReadTokens: 5},
|
||||
ContextLimit: sql.NullInt64{Int64: 4096, Valid: true},
|
||||
Runtime: 1500 * time.Millisecond,
|
||||
Content: []fantasy.Content{fantasy.TextContent{Text: "usage"}},
|
||||
Usage: fantasy.Usage{InputTokens: 100, OutputTokens: 20, TotalTokens: 120, ReasoningTokens: 3, CacheCreationTokens: 4, CacheReadTokens: 5},
|
||||
ContextLimit: sql.NullInt64{Int64: 4096, Valid: true},
|
||||
Runtime: 1500 * time.Millisecond,
|
||||
ProviderResponseID: "msg_usage",
|
||||
},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
@@ -256,6 +257,7 @@ func TestBuildCommitStepMessages_UsageRuntime(t *testing.T) {
|
||||
require.Equal(t, sql.NullInt64{Int64: 5, Valid: true}, msg.CacheReadTokens)
|
||||
require.Equal(t, sql.NullInt64{Int64: 4096, Valid: true}, msg.ContextLimit)
|
||||
require.Equal(t, sql.NullInt64{Int64: 1500, Valid: true}, msg.RuntimeMs)
|
||||
require.Equal(t, sql.NullString{String: "msg_usage", Valid: true}, msg.ProviderResponseID)
|
||||
}
|
||||
|
||||
func TestBuildCommitStepMessages_ToolTimestampsAndMCPConfigIDs(t *testing.T) {
|
||||
@@ -308,6 +310,7 @@ func TestBuildCompactionMessages_CompressedSummaryToolCallAndResult(t *testing.T
|
||||
ContextLimit: 1000,
|
||||
Runtime: 1500 * time.Millisecond,
|
||||
EstimatedContextTokens: 5,
|
||||
ProviderResponseID: "resp_summary",
|
||||
},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
@@ -319,11 +322,13 @@ func TestBuildCompactionMessages_CompressedSummaryToolCallAndResult(t *testing.T
|
||||
require.Equal(t, uuid.NullUUID{UUID: modelConfigID, Valid: true}, got.Messages[0].ModelConfigID)
|
||||
require.Equal(t, "system summary", parseMessageParts(t, got.Messages[0].Role, got.Messages[0].Content)[0].Text)
|
||||
require.False(t, got.Messages[0].RuntimeMs.Valid)
|
||||
require.False(t, got.Messages[0].ProviderResponseID.Valid)
|
||||
|
||||
require.Equal(t, database.ChatMessageRoleAssistant, got.Messages[1].Role)
|
||||
require.Equal(t, database.ChatMessageVisibilityUser, got.Messages[1].Visibility)
|
||||
require.True(t, got.Messages[1].Compressed)
|
||||
require.Equal(t, sql.NullInt64{Int64: 1500, Valid: true}, got.Messages[1].RuntimeMs)
|
||||
require.Equal(t, sql.NullString{String: "resp_summary", Valid: true}, got.Messages[1].ProviderResponseID)
|
||||
callPart := parseMessageParts(t, got.Messages[1].Role, got.Messages[1].Content)[0]
|
||||
require.Equal(t, codersdk.ChatMessagePartTypeToolCall, callPart.Type)
|
||||
require.Equal(t, "summary-1", callPart.ToolCallID)
|
||||
@@ -333,6 +338,7 @@ func TestBuildCompactionMessages_CompressedSummaryToolCallAndResult(t *testing.T
|
||||
require.Equal(t, database.ChatMessageVisibilityBoth, got.Messages[2].Visibility)
|
||||
require.True(t, got.Messages[2].Compressed)
|
||||
require.False(t, got.Messages[2].RuntimeMs.Valid)
|
||||
require.False(t, got.Messages[2].ProviderResponseID.Valid)
|
||||
resultPart := parseMessageParts(t, got.Messages[2].Role, got.Messages[2].Content)[0]
|
||||
require.Equal(t, codersdk.ChatMessagePartTypeToolResult, resultPart.Type)
|
||||
require.Equal(t, "summary-1", resultPart.ToolCallID)
|
||||
|
||||
Reference in New Issue
Block a user