mirror of
https://github.com/coder/coder.git
synced 2026-09-24 15:04:27 +08:00
feat: remove unused chat statuses pending, paused, and completed (#27064)
The chatd state machine only recognizes `waiting`, `running`, `error`, `requires_action`, and `interrupting`. Remove the unused `pending`, `paused`, and `completed` values from the database enum, backend, SDK, frontend, generated queries, and API docs. Migration `000543_chat_status_remove_unused` remaps existing `pending` rows to `running`, remaps `paused` and `completed` rows to `waiting`, drops the obsolete `idx_chats_pending` index, and recreates `chats_expanded` around the enum swap. It also removes the dead `AcquireChats` query and all remaining query literals for the deleted statuses. **NOTE**: The enum swap can break chat queries from older replicas during a mixed-version rollout because they still reference `'pending'::chat_status`. Chats are experimental, so this PR accepts that limited rollout window instead of adding a two-release expand and contract sequence. > This PR was authored by Mux (AI agent) on Mike's behalf.
This commit is contained in:
Generated
-6
@@ -17695,20 +17695,14 @@ const docTemplate = `{
|
||||
"type": "string",
|
||||
"enum": [
|
||||
"waiting",
|
||||
"pending",
|
||||
"running",
|
||||
"paused",
|
||||
"completed",
|
||||
"error",
|
||||
"requires_action",
|
||||
"interrupting"
|
||||
],
|
||||
"x-enum-varnames": [
|
||||
"ChatStatusWaiting",
|
||||
"ChatStatusPending",
|
||||
"ChatStatusRunning",
|
||||
"ChatStatusPaused",
|
||||
"ChatStatusCompleted",
|
||||
"ChatStatusError",
|
||||
"ChatStatusRequiresAction",
|
||||
"ChatStatusInterrupting"
|
||||
|
||||
Generated
-6
@@ -15944,20 +15944,14 @@
|
||||
"type": "string",
|
||||
"enum": [
|
||||
"waiting",
|
||||
"pending",
|
||||
"running",
|
||||
"paused",
|
||||
"completed",
|
||||
"error",
|
||||
"requires_action",
|
||||
"interrupting"
|
||||
],
|
||||
"x-enum-varnames": [
|
||||
"ChatStatusWaiting",
|
||||
"ChatStatusPending",
|
||||
"ChatStatusRunning",
|
||||
"ChatStatusPaused",
|
||||
"ChatStatusCompleted",
|
||||
"ChatStatusError",
|
||||
"ChatStatusRequiresAction",
|
||||
"ChatStatusInterrupting"
|
||||
|
||||
@@ -117,7 +117,7 @@ func waitForChatTerminalState(
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
return chat.Status != database.ChatStatusPending && chat.Status != database.ChatStatusRunning
|
||||
return chat.Status != database.ChatStatusRunning
|
||||
}, testutil.WaitLong, testutil.IntervalFast)
|
||||
}
|
||||
|
||||
|
||||
@@ -1669,15 +1669,6 @@ func scopedOrgRoleIdentifiers(names []string, orgID uuid.UUID) []rbac.RoleIdenti
|
||||
return out
|
||||
}
|
||||
|
||||
func (q *querier) AcquireChats(ctx context.Context, arg database.AcquireChatsParams) ([]database.Chat, error) {
|
||||
// AcquireChats is a system-level operation used by the chat processor.
|
||||
// Authorization is done at the system level, not per-user.
|
||||
if err := q.authorizeContext(ctx, policy.ActionUpdate, rbac.ResourceChat); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return q.db.AcquireChats(ctx, arg)
|
||||
}
|
||||
|
||||
func (q *querier) AcquireLock(ctx context.Context, id int64) error {
|
||||
return q.db.AcquireLock(ctx, id)
|
||||
}
|
||||
|
||||
@@ -546,16 +546,6 @@ func (s *MethodTestSuite) TestConnectionLogs() {
|
||||
}
|
||||
|
||||
func (s *MethodTestSuite) TestChats() {
|
||||
s.Run("AcquireChats", s.Mocked(func(dbm *dbmock.MockStore, faker *gofakeit.Faker, check *expects) {
|
||||
arg := database.AcquireChatsParams{
|
||||
StartedAt: dbtime.Now(),
|
||||
WorkerID: uuid.New(),
|
||||
NumChats: 1,
|
||||
}
|
||||
chat := testutil.Fake(s.T(), faker, database.Chat{})
|
||||
dbm.EXPECT().AcquireChats(gomock.Any(), arg).Return([]database.Chat{chat}, nil).AnyTimes()
|
||||
check.Args(arg).Asserts(rbac.ResourceChat, policy.ActionUpdate).Returns([]database.Chat{chat})
|
||||
}))
|
||||
s.Run("HydrateAgentChatsContext", s.Mocked(func(dbm *dbmock.MockStore, faker *gofakeit.Faker, check *expects) {
|
||||
arg := database.HydrateAgentChatsContextParams{AgentID: uuid.New()}
|
||||
dbm.EXPECT().HydrateAgentChatsContext(gomock.Any(), arg).Return(nil).AnyTimes()
|
||||
|
||||
-8
@@ -105,14 +105,6 @@ func (m queryMetricsStore) DeleteOrganization(ctx context.Context, id uuid.UUID)
|
||||
return r0
|
||||
}
|
||||
|
||||
func (m queryMetricsStore) AcquireChats(ctx context.Context, arg database.AcquireChatsParams) ([]database.Chat, error) {
|
||||
start := time.Now()
|
||||
r0, r1 := m.s.AcquireChats(ctx, arg)
|
||||
m.queryLatencies.WithLabelValues("AcquireChats").Observe(time.Since(start).Seconds())
|
||||
m.queryCounts.WithLabelValues(httpmw.ExtractHTTPRoute(ctx), httpmw.ExtractHTTPMethod(ctx), "AcquireChats").Inc()
|
||||
return r0, r1
|
||||
}
|
||||
|
||||
func (m queryMetricsStore) AcquireLock(ctx context.Context, pgAdvisoryXactLock int64) error {
|
||||
start := time.Now()
|
||||
r0 := m.s.AcquireLock(ctx, pgAdvisoryXactLock)
|
||||
|
||||
Generated
-15
@@ -45,21 +45,6 @@ func (m *MockStore) EXPECT() *MockStoreMockRecorder {
|
||||
return m.recorder
|
||||
}
|
||||
|
||||
// AcquireChats mocks base method.
|
||||
func (m *MockStore) AcquireChats(ctx context.Context, arg database.AcquireChatsParams) ([]database.Chat, error) {
|
||||
m.ctrl.T.Helper()
|
||||
ret := m.ctrl.Call(m, "AcquireChats", ctx, arg)
|
||||
ret0, _ := ret[0].([]database.Chat)
|
||||
ret1, _ := ret[1].(error)
|
||||
return ret0, ret1
|
||||
}
|
||||
|
||||
// AcquireChats indicates an expected call of AcquireChats.
|
||||
func (mr *MockStoreMockRecorder) AcquireChats(ctx, arg any) *gomock.Call {
|
||||
mr.mock.ctrl.T.Helper()
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AcquireChats", reflect.TypeOf((*MockStore)(nil).AcquireChats), ctx, arg)
|
||||
}
|
||||
|
||||
// AcquireLock mocks base method.
|
||||
func (m *MockStore) AcquireLock(ctx context.Context, pgAdvisoryXactLock int64) error {
|
||||
m.ctrl.T.Helper()
|
||||
|
||||
Generated
-5
@@ -362,10 +362,7 @@ CREATE TYPE chat_reasoning_effort AS ENUM (
|
||||
|
||||
CREATE TYPE chat_status AS ENUM (
|
||||
'waiting',
|
||||
'pending',
|
||||
'running',
|
||||
'paused',
|
||||
'completed',
|
||||
'error',
|
||||
'requires_action',
|
||||
'interrupting'
|
||||
@@ -4763,8 +4760,6 @@ CREATE INDEX idx_chats_owner ON chats USING btree (owner_id);
|
||||
|
||||
CREATE INDEX idx_chats_parent_chat_id ON chats USING btree (parent_chat_id);
|
||||
|
||||
CREATE INDEX idx_chats_pending ON chats USING btree (status) WHERE (status = 'pending'::chat_status);
|
||||
|
||||
CREATE INDEX idx_chats_root_chat_id ON chats USING btree (root_chat_id);
|
||||
|
||||
CREATE INDEX idx_chats_worker_acquisition_candidates ON chats USING btree (status, updated_at, id) WHERE (archived = false);
|
||||
|
||||
@@ -0,0 +1,5 @@
|
||||
-- No-op: the removed enum values are not restored, matching prior art such
|
||||
-- as 000377 and 000384. Restoring them would require another
|
||||
-- rename-create-cast-drop cycle, and the data cannot be restored anyway:
|
||||
-- rows remapped to 'running' or 'waiting' by the up migration keep their
|
||||
-- new status.
|
||||
@@ -0,0 +1,91 @@
|
||||
-- Remove legacy chat statuses that the chatd state machine treats as
|
||||
-- invalid. 'pending', 'paused', and 'completed' are never written by the
|
||||
-- backend anymore; the valid set is exactly what the state machine
|
||||
-- recognizes: waiting, running, error, requires_action, interrupting.
|
||||
|
||||
-- Remap any historical rows to the closest valid status. The column type
|
||||
-- is still the original chat_status here.
|
||||
--
|
||||
-- 'pending' meant queued work that no runner had picked up yet, so remap
|
||||
-- it to 'running': the worker acquisition query picks up 'running' chats
|
||||
-- without a worker and services them.
|
||||
UPDATE chats SET status = 'running'
|
||||
WHERE status = 'pending';
|
||||
|
||||
-- 'paused' and 'completed' were settled states; 'waiting' is the idle
|
||||
-- resting state and the column default.
|
||||
UPDATE chats SET status = 'waiting'
|
||||
WHERE status IN ('paused', 'completed');
|
||||
|
||||
-- The partial index's WHERE clause references 'pending', which is being
|
||||
-- removed. The index is obsolete now that the legacy AcquireChats query
|
||||
-- is gone.
|
||||
DROP INDEX idx_chats_pending;
|
||||
|
||||
-- The view selects c.status, so it must be dropped before the column's
|
||||
-- type can be altered. It is recreated verbatim below.
|
||||
DROP VIEW chats_expanded;
|
||||
|
||||
-- Recreate the enum without the removed values using the
|
||||
-- rename-create-cast-drop pattern.
|
||||
ALTER TYPE chat_status RENAME TO chat_status_old;
|
||||
CREATE TYPE chat_status AS ENUM (
|
||||
'waiting',
|
||||
'running',
|
||||
'error',
|
||||
'requires_action',
|
||||
'interrupting'
|
||||
);
|
||||
ALTER TABLE chats ALTER COLUMN status DROP DEFAULT;
|
||||
ALTER TABLE chats ALTER COLUMN status TYPE chat_status USING status::text::chat_status;
|
||||
ALTER TABLE chats ALTER COLUMN status SET DEFAULT 'waiting';
|
||||
DROP TYPE chat_status_old;
|
||||
|
||||
CREATE VIEW chats_expanded AS
|
||||
SELECT c.id,
|
||||
c.owner_id,
|
||||
c.workspace_id,
|
||||
c.title,
|
||||
c.status,
|
||||
c.worker_id,
|
||||
c.started_at,
|
||||
c.heartbeat_at,
|
||||
c.created_at,
|
||||
c.updated_at,
|
||||
c.parent_chat_id,
|
||||
c.root_chat_id,
|
||||
c.last_model_config_id,
|
||||
c.last_reasoning_effort,
|
||||
c.archived,
|
||||
c.last_error,
|
||||
c.mode,
|
||||
c.mcp_server_ids,
|
||||
c.labels,
|
||||
c.build_id,
|
||||
c.agent_id,
|
||||
c.pin_order,
|
||||
c.last_read_message_id,
|
||||
c.dynamic_tools,
|
||||
c.organization_id,
|
||||
c.plan_mode,
|
||||
c.client_type,
|
||||
c.last_turn_summary,
|
||||
c.snapshot_version,
|
||||
c.history_version,
|
||||
c.queue_version,
|
||||
c.generation_attempt,
|
||||
c.retry_state,
|
||||
c.retry_state_version,
|
||||
c.runner_id,
|
||||
c.requires_action_deadline_at,
|
||||
COALESCE(root.user_acl, c.user_acl) AS user_acl,
|
||||
COALESCE(root.group_acl, c.group_acl) AS group_acl,
|
||||
owner.username AS owner_username,
|
||||
owner.name AS owner_name,
|
||||
c.context_aggregate_hash,
|
||||
c.context_dirty_since,
|
||||
c.context_dirty_resources,
|
||||
c.context_error
|
||||
FROM ((chats c
|
||||
LEFT JOIN chats root ON ((root.id = COALESCE(c.root_chat_id, c.parent_chat_id))))
|
||||
JOIN visible_users owner ON ((owner.id = c.owner_id)));
|
||||
Generated
-9
@@ -1725,10 +1725,7 @@ type ChatStatus string
|
||||
|
||||
const (
|
||||
ChatStatusWaiting ChatStatus = "waiting"
|
||||
ChatStatusPending ChatStatus = "pending"
|
||||
ChatStatusRunning ChatStatus = "running"
|
||||
ChatStatusPaused ChatStatus = "paused"
|
||||
ChatStatusCompleted ChatStatus = "completed"
|
||||
ChatStatusError ChatStatus = "error"
|
||||
ChatStatusRequiresAction ChatStatus = "requires_action"
|
||||
ChatStatusInterrupting ChatStatus = "interrupting"
|
||||
@@ -1772,10 +1769,7 @@ func (ns NullChatStatus) Value() (driver.Value, error) {
|
||||
func (e ChatStatus) Valid() bool {
|
||||
switch e {
|
||||
case ChatStatusWaiting,
|
||||
ChatStatusPending,
|
||||
ChatStatusRunning,
|
||||
ChatStatusPaused,
|
||||
ChatStatusCompleted,
|
||||
ChatStatusError,
|
||||
ChatStatusRequiresAction,
|
||||
ChatStatusInterrupting:
|
||||
@@ -1787,10 +1781,7 @@ func (e ChatStatus) Valid() bool {
|
||||
func AllChatStatusValues() []ChatStatus {
|
||||
return []ChatStatus{
|
||||
ChatStatusWaiting,
|
||||
ChatStatusPending,
|
||||
ChatStatusRunning,
|
||||
ChatStatusPaused,
|
||||
ChatStatusCompleted,
|
||||
ChatStatusError,
|
||||
ChatStatusRequiresAction,
|
||||
ChatStatusInterrupting,
|
||||
|
||||
Generated
-3
@@ -13,9 +13,6 @@ import (
|
||||
)
|
||||
|
||||
type sqlcQuerier interface {
|
||||
// Acquires up to @num_chats pending chats for processing. Uses SKIP LOCKED
|
||||
// to prevent multiple replicas from acquiring the same chat.
|
||||
AcquireChats(ctx context.Context, arg AcquireChatsParams) ([]Chat, error)
|
||||
// Blocks until the lock is acquired.
|
||||
//
|
||||
// This must be called from within a transaction. The lock will be automatically
|
||||
|
||||
@@ -1280,11 +1280,11 @@ func TestChatContextHydration(t *testing.T) {
|
||||
hashH := []byte{0x01, 0x02, 0x03}
|
||||
hashOther := []byte{0xff, 0xee}
|
||||
|
||||
chatNull := newChat(database.ChatStatusWaiting, agent.ID) // never hydrated
|
||||
chatMatch := newChat(database.ChatStatusRunning, agent.ID) // already at hashH
|
||||
chatDrift := newChat(database.ChatStatusRunning, agent.ID) // drifted, active
|
||||
chatTerminal := newChat(database.ChatStatusCompleted, agent.ID) // drifted, terminal
|
||||
chatArchived := newChat(database.ChatStatusRunning, agent.ID) // drifted, archived
|
||||
chatNull := newChat(database.ChatStatusWaiting, agent.ID) // never hydrated
|
||||
chatMatch := newChat(database.ChatStatusRunning, agent.ID) // already at hashH
|
||||
chatDrift := newChat(database.ChatStatusRunning, agent.ID) // drifted, active
|
||||
chatTerminal := newChat(database.ChatStatusError, agent.ID) // drifted, terminal
|
||||
chatArchived := newChat(database.ChatStatusRunning, agent.ID) // drifted, archived
|
||||
chatOtherAgent := newChat(database.ChatStatusRunning, otherAgent.ID)
|
||||
|
||||
// Pin starting hashes; chatNull is intentionally left NULL.
|
||||
@@ -13019,7 +13019,7 @@ func TestChatPinOrderConstraints(t *testing.T) {
|
||||
|
||||
parent, err := db.InsertChat(ctx, database.InsertChatParams{
|
||||
OrganizationID: org.ID,
|
||||
Status: database.ChatStatusCompleted,
|
||||
Status: database.ChatStatusWaiting,
|
||||
ClientType: database.ChatClientTypeUi,
|
||||
OwnerID: owner.ID,
|
||||
LastModelConfigID: modelCfg.ID,
|
||||
@@ -13029,7 +13029,7 @@ func TestChatPinOrderConstraints(t *testing.T) {
|
||||
|
||||
child, err := db.InsertChat(ctx, database.InsertChatParams{
|
||||
OrganizationID: org.ID,
|
||||
Status: database.ChatStatusCompleted,
|
||||
Status: database.ChatStatusWaiting,
|
||||
ClientType: database.ChatClientTypeUi,
|
||||
OwnerID: owner.ID,
|
||||
LastModelConfigID: modelCfg.ID,
|
||||
@@ -13050,7 +13050,7 @@ func TestChatPinOrderConstraints(t *testing.T) {
|
||||
|
||||
chat, err := db.InsertChat(ctx, database.InsertChatParams{
|
||||
OrganizationID: org.ID,
|
||||
Status: database.ChatStatusCompleted,
|
||||
Status: database.ChatStatusWaiting,
|
||||
ClientType: database.ChatClientTypeUi,
|
||||
OwnerID: owner.ID,
|
||||
LastModelConfigID: modelCfg.ID,
|
||||
|
||||
Generated
+5
-167
@@ -5606,165 +5606,6 @@ func (q *sqlQuerier) UpdateChatModelConfig(ctx context.Context, arg UpdateChatMo
|
||||
return i, err
|
||||
}
|
||||
|
||||
const acquireChats = `-- name: AcquireChats :many
|
||||
WITH acquired_chats AS (
|
||||
UPDATE
|
||||
chats
|
||||
SET
|
||||
status = 'running'::chat_status,
|
||||
started_at = $1::timestamptz,
|
||||
heartbeat_at = $1::timestamptz,
|
||||
updated_at = $1::timestamptz,
|
||||
worker_id = $2::uuid
|
||||
WHERE
|
||||
id = ANY(
|
||||
SELECT
|
||||
id
|
||||
FROM
|
||||
chats
|
||||
WHERE
|
||||
status = 'pending'::chat_status
|
||||
AND archived = false
|
||||
ORDER BY
|
||||
updated_at ASC
|
||||
FOR UPDATE
|
||||
SKIP LOCKED
|
||||
LIMIT
|
||||
$3::int
|
||||
)
|
||||
RETURNING id, owner_id, workspace_id, title, status, worker_id, started_at, heartbeat_at, created_at, updated_at, parent_chat_id, root_chat_id, last_model_config_id, archived, last_error, mode, mcp_server_ids, labels, build_id, agent_id, pin_order, last_read_message_id, dynamic_tools, organization_id, plan_mode, client_type, last_turn_summary, user_acl, group_acl, snapshot_version, history_version, queue_version, generation_attempt, retry_state, retry_state_version, runner_id, requires_action_deadline_at, context_aggregate_hash, context_dirty_since, context_dirty_resources, context_error, last_reasoning_effort
|
||||
),
|
||||
chats_expanded AS (
|
||||
SELECT
|
||||
acquired_chats.id,
|
||||
acquired_chats.owner_id,
|
||||
acquired_chats.workspace_id,
|
||||
acquired_chats.title,
|
||||
acquired_chats.status,
|
||||
acquired_chats.worker_id,
|
||||
acquired_chats.started_at,
|
||||
acquired_chats.heartbeat_at,
|
||||
acquired_chats.created_at,
|
||||
acquired_chats.updated_at,
|
||||
acquired_chats.parent_chat_id,
|
||||
acquired_chats.root_chat_id,
|
||||
acquired_chats.last_model_config_id,
|
||||
acquired_chats.last_reasoning_effort,
|
||||
acquired_chats.archived,
|
||||
acquired_chats.last_error,
|
||||
acquired_chats.mode,
|
||||
acquired_chats.mcp_server_ids,
|
||||
acquired_chats.labels,
|
||||
acquired_chats.build_id,
|
||||
acquired_chats.agent_id,
|
||||
acquired_chats.pin_order,
|
||||
acquired_chats.last_read_message_id,
|
||||
acquired_chats.dynamic_tools,
|
||||
acquired_chats.organization_id,
|
||||
acquired_chats.plan_mode,
|
||||
acquired_chats.client_type,
|
||||
acquired_chats.last_turn_summary,
|
||||
acquired_chats.snapshot_version,
|
||||
acquired_chats.history_version,
|
||||
acquired_chats.queue_version,
|
||||
acquired_chats.generation_attempt,
|
||||
acquired_chats.retry_state,
|
||||
acquired_chats.retry_state_version,
|
||||
acquired_chats.runner_id,
|
||||
acquired_chats.requires_action_deadline_at,
|
||||
COALESCE(root.user_acl, acquired_chats.user_acl) AS user_acl,
|
||||
COALESCE(root.group_acl, acquired_chats.group_acl) AS group_acl,
|
||||
owner.username AS owner_username,
|
||||
owner.name AS owner_name,
|
||||
acquired_chats.context_aggregate_hash,
|
||||
acquired_chats.context_dirty_since,
|
||||
acquired_chats.context_dirty_resources,
|
||||
acquired_chats.context_error
|
||||
FROM
|
||||
acquired_chats
|
||||
LEFT JOIN chats root ON root.id = COALESCE(acquired_chats.root_chat_id, acquired_chats.parent_chat_id)
|
||||
JOIN visible_users owner ON owner.id = acquired_chats.owner_id
|
||||
)
|
||||
SELECT id, owner_id, workspace_id, title, status, worker_id, started_at, heartbeat_at, created_at, updated_at, parent_chat_id, root_chat_id, last_model_config_id, last_reasoning_effort, archived, last_error, mode, mcp_server_ids, labels, build_id, agent_id, pin_order, last_read_message_id, dynamic_tools, organization_id, plan_mode, client_type, last_turn_summary, snapshot_version, history_version, queue_version, generation_attempt, retry_state, retry_state_version, runner_id, requires_action_deadline_at, user_acl, group_acl, owner_username, owner_name, context_aggregate_hash, context_dirty_since, context_dirty_resources, context_error
|
||||
FROM chats_expanded
|
||||
`
|
||||
|
||||
type AcquireChatsParams struct {
|
||||
StartedAt time.Time `db:"started_at" json:"started_at"`
|
||||
WorkerID uuid.UUID `db:"worker_id" json:"worker_id"`
|
||||
NumChats int32 `db:"num_chats" json:"num_chats"`
|
||||
}
|
||||
|
||||
// Acquires up to @num_chats pending chats for processing. Uses SKIP LOCKED
|
||||
// to prevent multiple replicas from acquiring the same chat.
|
||||
func (q *sqlQuerier) AcquireChats(ctx context.Context, arg AcquireChatsParams) ([]Chat, error) {
|
||||
rows, err := q.db.QueryContext(ctx, acquireChats, arg.StartedAt, arg.WorkerID, arg.NumChats)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
var items []Chat
|
||||
for rows.Next() {
|
||||
var i Chat
|
||||
if err := rows.Scan(
|
||||
&i.ID,
|
||||
&i.OwnerID,
|
||||
&i.WorkspaceID,
|
||||
&i.Title,
|
||||
&i.Status,
|
||||
&i.WorkerID,
|
||||
&i.StartedAt,
|
||||
&i.HeartbeatAt,
|
||||
&i.CreatedAt,
|
||||
&i.UpdatedAt,
|
||||
&i.ParentChatID,
|
||||
&i.RootChatID,
|
||||
&i.LastModelConfigID,
|
||||
&i.LastReasoningEffort,
|
||||
&i.Archived,
|
||||
&i.LastError,
|
||||
&i.Mode,
|
||||
pq.Array(&i.MCPServerIDs),
|
||||
&i.Labels,
|
||||
&i.BuildID,
|
||||
&i.AgentID,
|
||||
&i.PinOrder,
|
||||
&i.LastReadMessageID,
|
||||
&i.DynamicTools,
|
||||
&i.OrganizationID,
|
||||
&i.PlanMode,
|
||||
&i.ClientType,
|
||||
&i.LastTurnSummary,
|
||||
&i.SnapshotVersion,
|
||||
&i.HistoryVersion,
|
||||
&i.QueueVersion,
|
||||
&i.GenerationAttempt,
|
||||
&i.RetryState,
|
||||
&i.RetryStateVersion,
|
||||
&i.RunnerID,
|
||||
&i.RequiresActionDeadlineAt,
|
||||
&i.UserACL,
|
||||
&i.GroupACL,
|
||||
&i.OwnerUsername,
|
||||
&i.OwnerName,
|
||||
&i.ContextAggregateHash,
|
||||
&i.ContextDirtySince,
|
||||
&i.ContextDirtyResources,
|
||||
&i.ContextError,
|
||||
); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
items = append(items, i)
|
||||
}
|
||||
if err := rows.Close(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return items, nil
|
||||
}
|
||||
|
||||
const acquireStaleChatDiffStatuses = `-- name: AcquireStaleChatDiffStatuses :many
|
||||
WITH acquired AS (
|
||||
UPDATE
|
||||
@@ -6036,7 +5877,7 @@ WITH to_archive AS (
|
||||
-- Redundant filter helps the planner use the partial index on created_at.
|
||||
AND c.created_at < $1::timestamptz
|
||||
-- New active statuses must be added here to prevent archiving.
|
||||
AND c.status NOT IN ('running', 'pending', 'paused', 'requires_action')
|
||||
AND c.status NOT IN ('running', 'requires_action')
|
||||
AND COALESCE(activity.last_activity_at, c.created_at) < $1::timestamptz
|
||||
-- Sorting by created_at lets Postgres drive the scan from the
|
||||
-- partial index instead of evaluating every LATERAL subquery
|
||||
@@ -6443,10 +6284,9 @@ SELECT id, owner_id, workspace_id, title, status, worker_id, started_at, heartbe
|
||||
FROM chats_expanded
|
||||
WHERE agent_id = $1::uuid
|
||||
AND archived = false
|
||||
-- Active statuses only: waiting, pending, running, paused,
|
||||
-- requires_action.
|
||||
-- Excludes completed and error (terminal states).
|
||||
AND status IN ('waiting', 'running', 'paused', 'pending', 'requires_action')
|
||||
-- Active statuses only: waiting, running, requires_action.
|
||||
-- Excludes error (terminal state) and interrupting.
|
||||
AND status IN ('waiting', 'running', 'requires_action')
|
||||
ORDER BY updated_at DESC
|
||||
`
|
||||
|
||||
@@ -6538,8 +6378,6 @@ WHERE
|
||||
AND chats_expanded.status NOT IN (
|
||||
'running'::chat_status,
|
||||
'interrupting'::chat_status,
|
||||
'pending'::chat_status,
|
||||
'paused'::chat_status,
|
||||
'requires_action'::chat_status
|
||||
)
|
||||
AND COALESCE(activity.last_activity_at, chats_expanded.created_at) < $1::timestamptz
|
||||
@@ -10400,7 +10238,7 @@ UPDATE chats
|
||||
SET context_dirty_since = $1
|
||||
WHERE agent_id = $2::uuid
|
||||
AND archived = false
|
||||
AND status IN ('waiting', 'running', 'paused', 'pending', 'requires_action')
|
||||
AND status IN ('waiting', 'running', 'requires_action')
|
||||
AND context_aggregate_hash IS NOT NULL
|
||||
AND context_aggregate_hash IS DISTINCT FROM $3
|
||||
AND context_dirty_since IS NULL
|
||||
|
||||
@@ -1469,7 +1469,7 @@ UPDATE chats
|
||||
SET context_dirty_since = @dirty_since
|
||||
WHERE agent_id = @agent_id::uuid
|
||||
AND archived = false
|
||||
AND status IN ('waiting', 'running', 'paused', 'pending', 'requires_action')
|
||||
AND status IN ('waiting', 'running', 'requires_action')
|
||||
AND context_aggregate_hash IS NOT NULL
|
||||
AND context_aggregate_hash IS DISTINCT FROM @aggregate_hash
|
||||
AND context_dirty_since IS NULL
|
||||
@@ -1540,90 +1540,6 @@ SELECT
|
||||
(SELECT COUNT(*)::int FROM genuinely_new) -
|
||||
(SELECT COUNT(*)::int FROM inserted) AS rejected_new_files;
|
||||
|
||||
-- name: AcquireChats :many
|
||||
-- Acquires up to @num_chats pending chats for processing. Uses SKIP LOCKED
|
||||
-- to prevent multiple replicas from acquiring the same chat.
|
||||
WITH acquired_chats AS (
|
||||
UPDATE
|
||||
chats
|
||||
SET
|
||||
status = 'running'::chat_status,
|
||||
started_at = @started_at::timestamptz,
|
||||
heartbeat_at = @started_at::timestamptz,
|
||||
updated_at = @started_at::timestamptz,
|
||||
worker_id = @worker_id::uuid
|
||||
WHERE
|
||||
id = ANY(
|
||||
SELECT
|
||||
id
|
||||
FROM
|
||||
chats
|
||||
WHERE
|
||||
status = 'pending'::chat_status
|
||||
AND archived = false
|
||||
ORDER BY
|
||||
updated_at ASC
|
||||
FOR UPDATE
|
||||
SKIP LOCKED
|
||||
LIMIT
|
||||
@num_chats::int
|
||||
)
|
||||
RETURNING *
|
||||
),
|
||||
chats_expanded AS (
|
||||
SELECT
|
||||
acquired_chats.id,
|
||||
acquired_chats.owner_id,
|
||||
acquired_chats.workspace_id,
|
||||
acquired_chats.title,
|
||||
acquired_chats.status,
|
||||
acquired_chats.worker_id,
|
||||
acquired_chats.started_at,
|
||||
acquired_chats.heartbeat_at,
|
||||
acquired_chats.created_at,
|
||||
acquired_chats.updated_at,
|
||||
acquired_chats.parent_chat_id,
|
||||
acquired_chats.root_chat_id,
|
||||
acquired_chats.last_model_config_id,
|
||||
acquired_chats.last_reasoning_effort,
|
||||
acquired_chats.archived,
|
||||
acquired_chats.last_error,
|
||||
acquired_chats.mode,
|
||||
acquired_chats.mcp_server_ids,
|
||||
acquired_chats.labels,
|
||||
acquired_chats.build_id,
|
||||
acquired_chats.agent_id,
|
||||
acquired_chats.pin_order,
|
||||
acquired_chats.last_read_message_id,
|
||||
acquired_chats.dynamic_tools,
|
||||
acquired_chats.organization_id,
|
||||
acquired_chats.plan_mode,
|
||||
acquired_chats.client_type,
|
||||
acquired_chats.last_turn_summary,
|
||||
acquired_chats.snapshot_version,
|
||||
acquired_chats.history_version,
|
||||
acquired_chats.queue_version,
|
||||
acquired_chats.generation_attempt,
|
||||
acquired_chats.retry_state,
|
||||
acquired_chats.retry_state_version,
|
||||
acquired_chats.runner_id,
|
||||
acquired_chats.requires_action_deadline_at,
|
||||
COALESCE(root.user_acl, acquired_chats.user_acl) AS user_acl,
|
||||
COALESCE(root.group_acl, acquired_chats.group_acl) AS group_acl,
|
||||
owner.username AS owner_username,
|
||||
owner.name AS owner_name,
|
||||
acquired_chats.context_aggregate_hash,
|
||||
acquired_chats.context_dirty_since,
|
||||
acquired_chats.context_dirty_resources,
|
||||
acquired_chats.context_error
|
||||
FROM
|
||||
acquired_chats
|
||||
LEFT JOIN chats root ON root.id = COALESCE(acquired_chats.root_chat_id, acquired_chats.parent_chat_id)
|
||||
JOIN visible_users owner ON owner.id = acquired_chats.owner_id
|
||||
)
|
||||
SELECT *
|
||||
FROM chats_expanded;
|
||||
|
||||
-- name: UpdateChatStatus :one
|
||||
WITH updated_chat AS (
|
||||
UPDATE
|
||||
@@ -2534,10 +2450,9 @@ SELECT *
|
||||
FROM chats_expanded
|
||||
WHERE agent_id = @agent_id::uuid
|
||||
AND archived = false
|
||||
-- Active statuses only: waiting, pending, running, paused,
|
||||
-- requires_action.
|
||||
-- Excludes completed and error (terminal states).
|
||||
AND status IN ('waiting', 'running', 'paused', 'pending', 'requires_action')
|
||||
-- Active statuses only: waiting, running, requires_action.
|
||||
-- Excludes error (terminal state) and interrupting.
|
||||
AND status IN ('waiting', 'running', 'requires_action')
|
||||
ORDER BY updated_at DESC;
|
||||
|
||||
-- name: SoftDeleteContextFileMessages :exec
|
||||
@@ -2630,8 +2545,6 @@ WHERE
|
||||
AND chats_expanded.status NOT IN (
|
||||
'running'::chat_status,
|
||||
'interrupting'::chat_status,
|
||||
'pending'::chat_status,
|
||||
'paused'::chat_status,
|
||||
'requires_action'::chat_status
|
||||
)
|
||||
AND COALESCE(activity.last_activity_at, chats_expanded.created_at) < @archive_cutoff::timestamptz
|
||||
@@ -2999,7 +2912,7 @@ WITH to_archive AS (
|
||||
-- Redundant filter helps the planner use the partial index on created_at.
|
||||
AND c.created_at < @archive_cutoff::timestamptz
|
||||
-- New active statuses must be added here to prevent archiving.
|
||||
AND c.status NOT IN ('running', 'pending', 'paused', 'requires_action')
|
||||
AND c.status NOT IN ('running', 'requires_action')
|
||||
AND COALESCE(activity.last_activity_at, c.created_at) < @archive_cutoff::timestamptz
|
||||
-- Sorting by created_at lets Postgres drive the scan from the
|
||||
-- partial index instead of evaluating every LATERAL subquery
|
||||
|
||||
+19
-111
@@ -1069,15 +1069,13 @@ func TestListChats(t *testing.T) {
|
||||
|
||||
require.Equal(t, firstUser.UserID, chat.OwnerID)
|
||||
require.Equal(t, modelConfig.ID, chat.LastModelConfigID)
|
||||
// The chat may have been picked up by the background
|
||||
// processor (via signalWake) before we list, so
|
||||
// accept any active status.
|
||||
// The chat may have been picked up by the chat worker
|
||||
// before we list, so accept any status it may have
|
||||
// reached by now.
|
||||
require.Contains(t, []codersdk.ChatStatus{
|
||||
codersdk.ChatStatusPending,
|
||||
codersdk.ChatStatusRunning,
|
||||
codersdk.ChatStatusError,
|
||||
codersdk.ChatStatusWaiting,
|
||||
codersdk.ChatStatusCompleted,
|
||||
}, chat.Status, "unexpected chat status: %s", chat.Status)
|
||||
require.NotZero(t, chat.CreatedAt)
|
||||
require.NotZero(t, chat.UpdatedAt)
|
||||
@@ -1100,9 +1098,8 @@ func TestListChats(t *testing.T) {
|
||||
// The list is already verified as sorted by UpdatedAt
|
||||
// descending (loop above). We intentionally do NOT
|
||||
// compare positions using the creation-time UpdatedAt
|
||||
// values because signalWake() may trigger background
|
||||
// processing that mutates UpdatedAt between CreateChat
|
||||
// and ListChats.
|
||||
// values because the chat worker may pick up a chat and
|
||||
// mutate UpdatedAt between CreateChat and ListChats.
|
||||
|
||||
memberChats, err := memberClient.ListChats(ctx, nil)
|
||||
require.NoError(t, err)
|
||||
@@ -1133,21 +1130,21 @@ func TestListChats(t *testing.T) {
|
||||
OwnerID: owner.ID,
|
||||
LastModelConfigID: modelConfig.ID,
|
||||
Title: "owner created chat",
|
||||
Status: database.ChatStatusCompleted,
|
||||
Status: database.ChatStatusWaiting,
|
||||
})
|
||||
sharedChat := dbgen.Chat(t, db, database.Chat{
|
||||
OrganizationID: firstUser.OrganizationID,
|
||||
OwnerID: member.ID,
|
||||
LastModelConfigID: modelConfig.ID,
|
||||
Title: "member shared chat",
|
||||
Status: database.ChatStatusCompleted,
|
||||
Status: database.ChatStatusWaiting,
|
||||
})
|
||||
unsharedReadableChat := dbgen.Chat(t, db, database.Chat{
|
||||
OrganizationID: firstUser.OrganizationID,
|
||||
OwnerID: firstUser.UserID,
|
||||
LastModelConfigID: modelConfig.ID,
|
||||
Title: "unshared readable chat",
|
||||
Status: database.ChatStatusCompleted,
|
||||
Status: database.ChatStatusWaiting,
|
||||
})
|
||||
|
||||
err := db.UpdateChatACLByID(dbauthz.As(ctx, rbac.Subject{
|
||||
@@ -1268,7 +1265,7 @@ func TestListChats(t *testing.T) {
|
||||
OwnerID: firstUser.UserID,
|
||||
LastModelConfigID: modelConfig.ID,
|
||||
Title: fmt.Sprintf("chat-%d", i),
|
||||
Status: database.ChatStatusCompleted,
|
||||
Status: database.ChatStatusWaiting,
|
||||
})
|
||||
createdChatIDs = append(createdChatIDs, dbChat.ID)
|
||||
}
|
||||
@@ -1351,7 +1348,7 @@ func TestListChats(t *testing.T) {
|
||||
OwnerID: firstUser.UserID,
|
||||
LastModelConfigID: modelConfig.ID,
|
||||
Title: "pinned-chat",
|
||||
Status: database.ChatStatusCompleted,
|
||||
Status: database.ChatStatusWaiting,
|
||||
})
|
||||
|
||||
// Fill page 1 with newer chats so the pinned chat
|
||||
@@ -1364,7 +1361,7 @@ func TestListChats(t *testing.T) {
|
||||
OwnerID: firstUser.UserID,
|
||||
LastModelConfigID: modelConfig.ID,
|
||||
Title: fmt.Sprintf("filler-%d", i),
|
||||
Status: database.ChatStatusCompleted,
|
||||
Status: database.ChatStatusWaiting,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -1411,7 +1408,7 @@ func TestListChats(t *testing.T) {
|
||||
OwnerID: firstUser.UserID,
|
||||
LastModelConfigID: modelConfig.ID,
|
||||
Title: fmt.Sprintf("cursor-pin-chat-%d", i),
|
||||
Status: database.ChatStatusCompleted,
|
||||
Status: database.ChatStatusWaiting,
|
||||
})
|
||||
createdChatIDs = append(createdChatIDs, dbChat.ID)
|
||||
}
|
||||
@@ -6552,7 +6549,7 @@ func TestChatPinOrder(t *testing.T) {
|
||||
OwnerID: firstUser.UserID,
|
||||
LastModelConfigID: modelConfig.ID,
|
||||
Title: "child chat",
|
||||
Status: database.ChatStatusCompleted,
|
||||
Status: database.ChatStatusWaiting,
|
||||
ParentChatID: uuid.NullUUID{UUID: parentChat.ID, Valid: true},
|
||||
RootChatID: uuid.NullUUID{UUID: parentChat.ID, Valid: true},
|
||||
})
|
||||
@@ -8659,8 +8656,7 @@ func TestPatchChatMessage(t *testing.T) {
|
||||
if getErr != nil {
|
||||
return false
|
||||
}
|
||||
return c.Status != codersdk.ChatStatusPending &&
|
||||
c.Status != codersdk.ChatStatusRunning
|
||||
return c.Status != codersdk.ChatStatusRunning
|
||||
}, testutil.IntervalFast, "initial chat processing did not finish")
|
||||
|
||||
messagesResult, err := client.GetChatMessages(ctx, chat.ID, nil)
|
||||
@@ -8696,8 +8692,7 @@ func TestPatchChatMessage(t *testing.T) {
|
||||
if getErr != nil {
|
||||
return false
|
||||
}
|
||||
return c.Status != codersdk.ChatStatusPending &&
|
||||
c.Status != codersdk.ChatStatusRunning
|
||||
return c.Status != codersdk.ChatStatusRunning
|
||||
}, testutil.IntervalFast, "post-edit chat processing did not finish")
|
||||
|
||||
updatedChat, err := client.GetChat(ctx, chat.ID)
|
||||
@@ -8997,7 +8992,7 @@ func TestRegenerateChatTitle(t *testing.T) {
|
||||
|
||||
_, err = db.UpdateChatStatus(dbauthz.AsSystemRestricted(ctx), database.UpdateChatStatusParams{
|
||||
ID: chat.ID,
|
||||
Status: database.ChatStatusCompleted,
|
||||
Status: database.ChatStatusWaiting,
|
||||
WorkerID: uuid.NullUUID{},
|
||||
StartedAt: sql.NullTime{},
|
||||
HeartbeatAt: sql.NullTime{},
|
||||
@@ -9020,47 +9015,6 @@ func TestRegenerateChatTitle(t *testing.T) {
|
||||
)
|
||||
})
|
||||
|
||||
t.Run("PendingWithoutWorker", func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
ctx := testutil.Context(t, testutil.WaitLong)
|
||||
client, db := newChatClientWithDatabase(t)
|
||||
user := coderdtest.CreateFirstUser(t, client.Client)
|
||||
modelConfig := createTitleGenerationModelConfig(t, client)
|
||||
|
||||
chat := dbgen.Chat(t, db, database.Chat{
|
||||
OrganizationID: user.OrganizationID,
|
||||
OwnerID: user.UserID,
|
||||
LastModelConfigID: modelConfig.ID,
|
||||
Title: "pending chat without worker",
|
||||
})
|
||||
seedManualTitleSourceMessage(t, db, chat, modelConfig.ID)
|
||||
|
||||
var err error
|
||||
chat, err = db.UpdateChatStatus(dbauthz.AsSystemRestricted(ctx), database.UpdateChatStatusParams{
|
||||
ID: chat.ID,
|
||||
Status: database.ChatStatusPending,
|
||||
WorkerID: uuid.NullUUID{},
|
||||
StartedAt: sql.NullTime{},
|
||||
HeartbeatAt: sql.NullTime{},
|
||||
LastError: pqtype.NullRawMessage{},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
// Pending chats are never acquired
|
||||
// (GetChatWorkerAcquisitionCandidates excludes the status), so
|
||||
// manual title regeneration must still proceed.
|
||||
updated, err := client.RegenerateChatTitle(ctx, chat.ID)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, "Test Chat", updated.Title)
|
||||
|
||||
persisted, err := db.GetChatByID(dbauthz.AsSystemRestricted(ctx), chat.ID)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, "Test Chat", persisted.Title)
|
||||
require.Equal(t, database.ChatStatusPending, persisted.Status)
|
||||
require.False(t, persisted.WorkerID.Valid)
|
||||
})
|
||||
|
||||
t.Run("PasteOnlyChat", func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
@@ -9074,7 +9028,7 @@ func TestRegenerateChatTitle(t *testing.T) {
|
||||
OwnerID: user.UserID,
|
||||
LastModelConfigID: modelConfig.ID,
|
||||
Title: "New Chat",
|
||||
Status: database.ChatStatusCompleted,
|
||||
Status: database.ChatStatusWaiting,
|
||||
})
|
||||
// The chat's only user message is a synthetic pasted-text
|
||||
// attachment with no text parts.
|
||||
@@ -9158,7 +9112,7 @@ func TestRegenerateChatTitle(t *testing.T) {
|
||||
|
||||
_, err = db.UpdateChatStatus(dbauthz.AsSystemRestricted(ctx), database.UpdateChatStatusParams{
|
||||
ID: chat.ID,
|
||||
Status: database.ChatStatusCompleted,
|
||||
Status: database.ChatStatusWaiting,
|
||||
WorkerID: uuid.NullUUID{},
|
||||
StartedAt: sql.NullTime{},
|
||||
HeartbeatAt: sql.NullTime{},
|
||||
@@ -9247,52 +9201,6 @@ func TestProposeChatTitle(t *testing.T) {
|
||||
requireSDKError(t, err, http.StatusUnauthorized)
|
||||
})
|
||||
|
||||
t.Run("PendingWithoutWorker", func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
ctx := testutil.Context(t, testutil.WaitLong)
|
||||
client, db := newChatClientWithDatabase(t)
|
||||
user := coderdtest.CreateFirstUser(t, client.Client)
|
||||
modelConfig := createTitleGenerationModelConfig(t, client)
|
||||
|
||||
chat := dbgen.Chat(t, db, database.Chat{
|
||||
OrganizationID: user.OrganizationID,
|
||||
OwnerID: user.UserID,
|
||||
LastModelConfigID: modelConfig.ID,
|
||||
Title: "pending chat without worker",
|
||||
})
|
||||
seedManualTitleSourceMessage(t, db, chat, modelConfig.ID)
|
||||
|
||||
var err error
|
||||
chat, err = db.UpdateChatStatus(dbauthz.AsSystemRestricted(ctx), database.UpdateChatStatusParams{
|
||||
ID: chat.ID,
|
||||
Status: database.ChatStatusPending,
|
||||
WorkerID: uuid.NullUUID{},
|
||||
StartedAt: sql.NullTime{},
|
||||
HeartbeatAt: sql.NullTime{},
|
||||
LastError: pqtype.NullRawMessage{},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
before, err := db.GetChatByID(dbauthz.AsSystemRestricted(ctx), chat.ID)
|
||||
require.NoError(t, err)
|
||||
|
||||
// Pending chats are never acquired
|
||||
// (GetChatWorkerAcquisitionCandidates excludes the status), so
|
||||
// title proposal must still proceed.
|
||||
resp, err := client.ProposeChatTitle(ctx, chat.ID)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, "Test Chat", resp.Title)
|
||||
|
||||
persisted, err := db.GetChatByID(dbauthz.AsSystemRestricted(ctx), chat.ID)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, before.Title, persisted.Title,
|
||||
"propose must not persist the suggested title")
|
||||
require.Equal(t, database.ChatStatusPending, persisted.Status)
|
||||
require.False(t, persisted.WorkerID.Valid)
|
||||
require.True(t, persisted.UpdatedAt.Equal(before.UpdatedAt))
|
||||
})
|
||||
|
||||
t.Run("DoesNotBumpHistoryVersion", func(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
@@ -9467,7 +9375,7 @@ func TestManualTitleEndpointsPassCallerAPIKeyToAIGateway(t *testing.T) {
|
||||
OwnerID: firstUser.UserID,
|
||||
LastModelConfigID: modelConfig.ID,
|
||||
Title: "initial title",
|
||||
Status: database.ChatStatusCompleted,
|
||||
Status: database.ChatStatusWaiting,
|
||||
})
|
||||
content, err := chatprompt.MarshalParts([]codersdk.ChatMessagePart{
|
||||
codersdk.ChatMessageText("manual title source"),
|
||||
|
||||
@@ -1681,7 +1681,7 @@ func TestChatsTelemetry(t *testing.T) {
|
||||
OwnerID: user.ID,
|
||||
LastModelConfigID: modelCfg2.ID,
|
||||
Title: "Child Chat",
|
||||
Status: database.ChatStatusCompleted,
|
||||
Status: database.ChatStatusWaiting,
|
||||
ParentChatID: uuid.NullUUID{UUID: rootChat.ID, Valid: true},
|
||||
RootChatID: uuid.NullUUID{UUID: rootChat.ID, Valid: true},
|
||||
})
|
||||
@@ -1852,7 +1852,7 @@ func TestChatsTelemetry(t *testing.T) {
|
||||
require.NotNil(t, foundChild.RootChatID)
|
||||
assert.Equal(t, rootChat.ID, *foundChild.RootChatID)
|
||||
assert.Nil(t, foundChild.WorkspaceID)
|
||||
assert.Equal(t, "completed", foundChild.Status)
|
||||
assert.Equal(t, "waiting", foundChild.Status)
|
||||
assert.Equal(t, modelCfg2.ID, foundChild.LastModelConfigID)
|
||||
assert.Nil(t, foundChild.Mode)
|
||||
assert.False(t, foundChild.Archived)
|
||||
@@ -1969,7 +1969,7 @@ func TestChatDiffStatusSummaryTelemetry(t *testing.T) {
|
||||
OwnerID: user.ID,
|
||||
LastModelConfigID: modelCfg.ID,
|
||||
Title: "Chat " + state,
|
||||
Status: database.ChatStatusCompleted,
|
||||
Status: database.ChatStatusWaiting,
|
||||
})
|
||||
now := dbtime.Now()
|
||||
_, chatErr := db.UpsertChatDiffStatus(ctx, database.UpsertChatDiffStatusParams{
|
||||
|
||||
@@ -4366,9 +4366,6 @@ func (p *Server) maybeFinalizeTurnStatusLabelAndPush(
|
||||
case database.ChatStatusWaiting:
|
||||
p.finalizeSuccessfulTurnStatusLabelAndPush(ctx, chat, status, runResult, logger)
|
||||
|
||||
case database.ChatStatusPending:
|
||||
p.setLastTurnSummaryAsync(ctx, chat, fallbackTurnStatusLabel(status), logger)
|
||||
|
||||
case database.ChatStatusError:
|
||||
p.clearLastTurnSummaryAsync(ctx, chat, logger)
|
||||
if p.webpushConfigured() {
|
||||
|
||||
@@ -1817,7 +1817,7 @@ func TestSendMessageRejectsInvalidQueuedModelConfigID(t *testing.T) {
|
||||
|
||||
chat := dbgen.Chat(t, db, database.Chat{
|
||||
OrganizationID: org.ID,
|
||||
Status: database.ChatStatusPending,
|
||||
Status: database.ChatStatusRunning,
|
||||
OwnerID: user.ID,
|
||||
LastModelConfigID: modelConfig.ID,
|
||||
Title: "reject invalid queued model config",
|
||||
@@ -2934,17 +2934,17 @@ func TestUpdateChatStatusPersistsLastError(t *testing.T) {
|
||||
require.Equal(t, wantPayload, requireChatLastErrorPayload(t, fromDB.LastError))
|
||||
|
||||
// Verify the error is cleared when the chat transitions to a
|
||||
// non-error status (e.g. pending after a retry).
|
||||
// non-error status (e.g. running after a retry).
|
||||
chat, err = db.UpdateChatStatus(ctx, database.UpdateChatStatusParams{
|
||||
ID: chat.ID,
|
||||
Status: database.ChatStatusPending,
|
||||
Status: database.ChatStatusRunning,
|
||||
WorkerID: uuid.NullUUID{},
|
||||
StartedAt: sql.NullTime{},
|
||||
HeartbeatAt: sql.NullTime{},
|
||||
LastError: pqtype.NullRawMessage{},
|
||||
})
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, database.ChatStatusPending, chat.Status)
|
||||
require.Equal(t, database.ChatStatusRunning, chat.Status)
|
||||
require.False(t, chat.LastError.Valid)
|
||||
|
||||
fromDB, err = db.GetChatByID(ctx, chat.ID)
|
||||
@@ -5250,16 +5250,10 @@ func TestHeartbeatNoWorkspaceNoBump(t *testing.T) {
|
||||
require.Equal(t, 0, count, "expected no workspaces to be flushed when chat has no workspace")
|
||||
}
|
||||
|
||||
// waitForChatProcessed waits for a wake-triggered processOnce to
|
||||
// fully complete for the given chat. It polls until the chat leaves
|
||||
// both pending and running states (meaning processChat has finished
|
||||
// its cleanup and updated the DB), then calls WaitUntilIdleForTest.
|
||||
//
|
||||
// Waiting for a terminal state (not just "not pending") avoids a
|
||||
// WaitGroup Add/Wait race: AcquireChats changes the DB status to
|
||||
// running before processOnce calls inflight.Add(1). If we only
|
||||
// waited for status != pending, we could call Wait() while Add(1)
|
||||
// hasn't happened yet.
|
||||
// waitForChatProcessed waits for the chat worker to fully handle a
|
||||
// wake for the given chat. It polls until the chat leaves the running
|
||||
// state (the worker has finished the turn and updated the DB), then
|
||||
// calls WaitUntilIdleForTest so tracked background work settles.
|
||||
func waitForChatProcessed(
|
||||
ctx context.Context,
|
||||
t *testing.T,
|
||||
@@ -5273,18 +5267,18 @@ func waitForChatProcessed(
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
// Wait until the chat reaches a terminal state. Neither
|
||||
// pending (waiting to be acquired) nor running (being
|
||||
// processed). This guarantees that inflight.Add(1) has
|
||||
// already been called by processOnce.
|
||||
return c.Status != database.ChatStatusPending &&
|
||||
c.Status != database.ChatStatusRunning
|
||||
// Wait until the chat reaches a settled state (not being
|
||||
// processed). This guarantees the wake was picked up and its
|
||||
// background work registered before WaitUntilIdleForTest
|
||||
// checks for idleness.
|
||||
return c.Status != database.ChatStatusRunning
|
||||
}, testutil.WaitShort, testutil.IntervalFast)
|
||||
chatd.WaitUntilIdleForTest(server)
|
||||
}
|
||||
|
||||
// newTestServer creates a passive server that never calls
|
||||
// processOnce on its own.
|
||||
// newTestServer creates a passive server whose periodic chat
|
||||
// acquisition is effectively disabled, so chats are only processed in
|
||||
// response to explicit wakes.
|
||||
func newTestServer(
|
||||
t *testing.T,
|
||||
db database.Store,
|
||||
@@ -8405,7 +8399,7 @@ func TestProposeChatTitle_DebugRun(t *testing.T) {
|
||||
|
||||
chat := dbgen.Chat(t, db, database.Chat{
|
||||
OrganizationID: org.ID,
|
||||
Status: database.ChatStatusCompleted,
|
||||
Status: database.ChatStatusWaiting,
|
||||
ClientType: database.ChatClientTypeUi,
|
||||
OwnerID: user.ID,
|
||||
Title: "original title",
|
||||
@@ -11823,54 +11817,6 @@ func TestPromoteQueuedPreservesReasoningEffort(t *testing.T) {
|
||||
require.Equal(t, database.ChatReasoningEffortHigh, storedChat.LastReasoningEffort.ChatReasoningEffort)
|
||||
}
|
||||
|
||||
// TestPromoteQueuedWhileRequiresActionMixedTools guards against
|
||||
func TestAcquireChatsSkipsArchivedPendingChat(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
db, ps := dbtestutil.NewDB(t)
|
||||
_ = newTestServer(t, db, ps, uuid.New())
|
||||
|
||||
ctx := testutil.Context(t, testutil.WaitLong)
|
||||
user, org, model := seedChatDependencies(t, db)
|
||||
|
||||
archivedChat := dbgen.Chat(t, db, database.Chat{
|
||||
OwnerID: user.ID,
|
||||
OrganizationID: org.ID,
|
||||
Title: "acquire-skip-archived",
|
||||
LastModelConfigID: model.ID,
|
||||
})
|
||||
|
||||
// Archive the chat, then force it to pending.
|
||||
_, err := db.ArchiveChatByID(ctx, archivedChat.ID)
|
||||
require.NoError(t, err)
|
||||
|
||||
_, err = db.UpdateChatStatus(ctx, database.UpdateChatStatusParams{
|
||||
ID: archivedChat.ID,
|
||||
Status: database.ChatStatusPending,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
// Insert a second, non-archived pending chat so the result
|
||||
// slice is non-empty and the assertion is not vacuously true.
|
||||
activeChat := dbgen.Chat(t, db, database.Chat{
|
||||
OwnerID: user.ID,
|
||||
OrganizationID: org.ID,
|
||||
Title: "acquire-active",
|
||||
LastModelConfigID: model.ID,
|
||||
Status: database.ChatStatusPending,
|
||||
})
|
||||
|
||||
now := time.Now()
|
||||
acquired, err := db.AcquireChats(ctx, database.AcquireChatsParams{
|
||||
WorkerID: uuid.New(),
|
||||
StartedAt: now,
|
||||
NumChats: 10,
|
||||
})
|
||||
require.NoError(t, err)
|
||||
require.Len(t, acquired, 1, "only the non-archived chat should be acquired")
|
||||
require.Equal(t, activeChat.ID, acquired[0].ID)
|
||||
}
|
||||
|
||||
// TestAdvisorGating_ExperimentDisabled verifies that the advisor tool is
|
||||
// not attached when the chat-advisor experiment is absent from the
|
||||
// experiments list, even if the DB-stored advisor config has Enabled=true.
|
||||
|
||||
@@ -113,9 +113,8 @@ func (s ExecutionState) QueueNonEmpty() bool {
|
||||
//
|
||||
// The classifier is a single flat switch over the valid (status,
|
||||
// archived, queue) tuples in the chat execution state model. Anything
|
||||
// outside that set (legacy pending/paused/completed statuses, archived
|
||||
// busy states, waiting with a non-empty queue, future enum values)
|
||||
// falls through to [StateInvalid].
|
||||
// outside that set (archived busy states, waiting with a non-empty
|
||||
// queue, future enum values) falls through to [StateInvalid].
|
||||
//
|
||||
//nolint:revive // queueNonEmpty/exists are simple classifier inputs.
|
||||
func ClassifyExecutionState(chat database.Chat, queueNonEmpty, exists bool) ExecutionState {
|
||||
|
||||
@@ -243,8 +243,7 @@ func waitForChatDone(
|
||||
if event.Status != nil {
|
||||
t.Logf("[%s] status → %s", label, event.Status.Status)
|
||||
switch event.Status.Status {
|
||||
case codersdk.ChatStatusWaiting,
|
||||
codersdk.ChatStatusCompleted:
|
||||
case codersdk.ChatStatusWaiting:
|
||||
return
|
||||
case codersdk.ChatStatusError:
|
||||
require.FailNow(t, label+" ended with error status")
|
||||
|
||||
@@ -1087,8 +1087,6 @@ func turnStatusLabelStateContext(status database.ChatStatus) string {
|
||||
switch status {
|
||||
case database.ChatStatusWaiting:
|
||||
return "The turn finished and the chat is idle."
|
||||
case database.ChatStatusPending:
|
||||
return "Another user message is queued and the chat will continue."
|
||||
case database.ChatStatusRequiresAction:
|
||||
return "The chat is waiting for user input or action."
|
||||
case database.ChatStatusError:
|
||||
@@ -1102,8 +1100,6 @@ func fallbackTurnStatusLabel(status database.ChatStatus) string {
|
||||
switch status {
|
||||
case database.ChatStatusWaiting:
|
||||
return "Finished latest turn"
|
||||
case database.ChatStatusPending:
|
||||
return "Still working on request"
|
||||
case database.ChatStatusRequiresAction:
|
||||
return "Waiting for user input"
|
||||
case database.ChatStatusError:
|
||||
|
||||
@@ -863,7 +863,6 @@ func TestFallbackTurnStatusLabel(t *testing.T) {
|
||||
want string
|
||||
}{
|
||||
{status: database.ChatStatusWaiting, want: "Finished latest turn"},
|
||||
{status: database.ChatStatusPending, want: "Still working on request"},
|
||||
{status: database.ChatStatusRequiresAction, want: "Waiting for user input"},
|
||||
{status: database.ChatStatusError, want: "Hit an error"},
|
||||
{status: database.ChatStatus("unknown"), want: "Updated chat status"},
|
||||
|
||||
@@ -105,7 +105,7 @@ func createComputerUseParentChild(
|
||||
AgentID: uuid.NullUUID{UUID: agent.ID, Valid: true},
|
||||
LastModelConfigID: model.ID,
|
||||
Title: parentTitle,
|
||||
Status: database.ChatStatusPending,
|
||||
Status: database.ChatStatusRunning,
|
||||
})
|
||||
|
||||
// Insert the child chat directly via DB to avoid triggering
|
||||
@@ -121,7 +121,7 @@ func createComputerUseParentChild(
|
||||
LastModelConfigID: model.ID,
|
||||
Title: childTitle,
|
||||
Mode: database.NullChatMode{ChatMode: database.ChatModeComputerUse, Valid: true},
|
||||
Status: database.ChatStatusPending,
|
||||
Status: database.ChatStatusRunning,
|
||||
})
|
||||
|
||||
return parent, child
|
||||
|
||||
@@ -800,8 +800,7 @@ func (p *Server) subagentTools(
|
||||
|
||||
interrupted := false
|
||||
if args.Interrupt && targetChatInfo != nil {
|
||||
interrupted = targetChatInfo.Status == database.ChatStatusRunning ||
|
||||
targetChatInfo.Status == database.ChatStatusPending
|
||||
interrupted = targetChatInfo.Status == database.ChatStatusRunning
|
||||
}
|
||||
return toolJSONResponse(withSubagentType(map[string]any{
|
||||
"chat_id": targetChat.ID.String(),
|
||||
@@ -862,8 +861,8 @@ func (p *Server) subagentTools(
|
||||
"sort order is best-effort: an agent's position may shift "+
|
||||
"if its updated_at changes between calls. Each "+
|
||||
"agent has chat_id, title, type, status, created_at, "+
|
||||
"updated_at. Status: pending/running = working, "+
|
||||
"interrupting = transient, waiting/completed = idle, "+
|
||||
"updated_at. Status: running = working, "+
|
||||
"interrupting = transient, waiting = idle, "+
|
||||
"error = stopped on error.",
|
||||
func(ctx context.Context, args listAgentsArgs, _ fantasy.ToolCall) (fantasy.ToolResponse, error) {
|
||||
if currentChat == nil {
|
||||
@@ -1434,8 +1433,7 @@ func (p *Server) checkSubagentCompletion(
|
||||
// waiting (no queued messages) or running (queued messages).
|
||||
// Treat it as not-done so the agent settles before
|
||||
// classification, avoiding stale partial output.
|
||||
if chat.Status == database.ChatStatusPending ||
|
||||
chat.Status == database.ChatStatusRunning ||
|
||||
if chat.Status == database.ChatStatusRunning ||
|
||||
chat.Status == database.ChatStatusInterrupting {
|
||||
return chat, "", false, nil
|
||||
}
|
||||
|
||||
@@ -304,8 +304,8 @@ func buildSpawnAgentDescription(
|
||||
"After spawning, use wait_agent to retrieve the result. Agents persist " +
|
||||
"after completion; reuse an agent via message_agent for follow-up work " +
|
||||
"when it already has relevant context. Spawned agents are your " +
|
||||
"responsibility: do not abandon one in a working state (pending or " +
|
||||
"running); retrieve its result, redirect it with message_agent, or stop " +
|
||||
"responsibility: do not abandon one in a working state (running); " +
|
||||
"retrieve its result, redirect it with message_agent, or stop " +
|
||||
"it with interrupt_agent."
|
||||
if currentChat.PlanMode.Valid && currentChat.PlanMode.ChatPlanMode == database.ChatPlanModePlan {
|
||||
description += " During plan mode, type=\"" + subagentTypeGeneral +
|
||||
|
||||
@@ -7,7 +7,6 @@ import (
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
|
||||
"charm.land/fantasy"
|
||||
"github.com/google/uuid"
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
@@ -17,7 +16,6 @@ import (
|
||||
"github.com/coder/coder/v2/coderd/database/dbtestutil"
|
||||
"github.com/coder/coder/v2/coderd/x/chatd/chatprompt"
|
||||
"github.com/coder/coder/v2/coderd/x/chatd/chatstate"
|
||||
"github.com/coder/coder/v2/coderd/x/chatd/chattest"
|
||||
"github.com/coder/coder/v2/codersdk"
|
||||
"github.com/coder/coder/v2/testutil"
|
||||
)
|
||||
@@ -116,89 +114,6 @@ func TestUpdateLastTurnSummaryRejectsStaleWrites(t *testing.T) {
|
||||
require.Equal(t, sql.NullString{String: "fresh summary", Valid: true}, fetched.LastTurnSummary)
|
||||
}
|
||||
|
||||
func TestPendingChatPersistsSummaryButSkipsWebPush(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
db, ps := dbtestutil.NewDB(t)
|
||||
ctx := testutil.Context(t, testutil.WaitMedium)
|
||||
owner := dbgen.User(t, db, database.User{})
|
||||
org := dbgen.Organization(t, db, database.Organization{})
|
||||
dbgen.OrganizationMember(t, db, database.OrganizationMember{
|
||||
UserID: owner.ID,
|
||||
OrganizationID: org.ID,
|
||||
})
|
||||
|
||||
provider := dbgen.ChatProvider(t, db, database.ChatProvider{
|
||||
Provider: "openai",
|
||||
DisplayName: "OpenAI",
|
||||
APIKey: "test-key",
|
||||
Enabled: true,
|
||||
})
|
||||
|
||||
modelCfg, err := db.InsertChatModelConfig(ctx, database.InsertChatModelConfigParams{
|
||||
AIProviderID: uuid.NullUUID{UUID: provider.ID, Valid: true},
|
||||
Model: "test-model",
|
||||
DisplayName: "Test Model",
|
||||
CreatedBy: uuid.NullUUID{UUID: owner.ID, Valid: true},
|
||||
UpdatedBy: uuid.NullUUID{UUID: owner.ID, Valid: true},
|
||||
Enabled: true,
|
||||
IsDefault: true,
|
||||
ContextLimit: 128000,
|
||||
CompressionThreshold: 80,
|
||||
Options: json.RawMessage(`{}`),
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
chat, err := db.InsertChat(ctx, database.InsertChatParams{
|
||||
OrganizationID: org.ID,
|
||||
Status: database.ChatStatusPending,
|
||||
ClientType: database.ChatClientTypeUi,
|
||||
OwnerID: owner.ID,
|
||||
LastModelConfigID: modelCfg.ID,
|
||||
Title: "summary-pending-chat",
|
||||
})
|
||||
require.NoError(t, err)
|
||||
|
||||
const summary = "Still working on request"
|
||||
var generateCalls atomic.Int32
|
||||
model := &chattest.FakeModel{
|
||||
ProviderName: "openai",
|
||||
ModelName: "test-model",
|
||||
GenerateFn: func(_ context.Context, _ fantasy.Call) (*fantasy.Response, error) {
|
||||
generateCalls.Add(1)
|
||||
return &fantasy.Response{
|
||||
Content: fantasy.ResponseContent{
|
||||
fantasy.TextContent{Text: "Unexpected label"},
|
||||
},
|
||||
}, nil
|
||||
},
|
||||
}
|
||||
|
||||
dispatcher := &recordingWebpushDispatcher{}
|
||||
logger := slogtest.Make(t, &slogtest.Options{IgnoreErrors: true})
|
||||
server := &Server{ctx: t.Context(), db: db, pubsub: ps, webpushDispatcher: dispatcher}
|
||||
server.maybeFinalizeTurnStatusLabelAndPush(
|
||||
context.WithoutCancel(ctx),
|
||||
chat,
|
||||
database.ChatStatusPending,
|
||||
"",
|
||||
runChatResult{
|
||||
FinalAssistantText: "I finished the queued turn.",
|
||||
StatusLabelModel: model,
|
||||
FallbackProvider: model.Provider(),
|
||||
FallbackModel: model.Model(),
|
||||
},
|
||||
logger,
|
||||
)
|
||||
server.drainInflight()
|
||||
|
||||
fetched, err := db.GetChatByID(ctx, chat.ID)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, sql.NullString{String: summary, Valid: true}, fetched.LastTurnSummary)
|
||||
require.Equal(t, int32(0), generateCalls.Load())
|
||||
require.Equal(t, int32(0), dispatcher.dispatchCount.Load())
|
||||
}
|
||||
|
||||
func TestSuccessfulChildChatOutcomeSkipsSummaryAndWebPush(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
|
||||
Reference in New Issue
Block a user