diff --git a/apps/backend/internal/task/repository/sqlite/pending_interactions.go b/apps/backend/internal/task/repository/sqlite/pending_interactions.go index f6ab17509a..f2ebdc53ad 100644 --- a/apps/backend/internal/task/repository/sqlite/pending_interactions.go +++ b/apps/backend/internal/task/repository/sqlite/pending_interactions.go @@ -2,7 +2,6 @@ package sqlite import ( "context" - "fmt" "strings" "github.com/kandev/kandev/internal/agentctl/tracing" @@ -62,25 +61,21 @@ func buildPendingInteractionQuery( pendingIDExpr := dialect.JSONExtract(driverName, "m.metadata", "pending_id") scopedSessions, args := pendingInteractionSessionScope(filter) kindConditions, kindArgs := pendingInteractionKindClauses(filter.Kinds) + predicate, orderBy := currentTurnAuthority(driverName, "turn_row") - query := fmt.Sprintf( - pendingInteractionQueryTemplate, - scopedSessions, - turnAuthorityPredicate(driverName, "turn_row"), - nonTerminalSessionPredicate("turn_row"), - pendingIDExpr, - models.MessageTypeClarificationRequest, - statusExpr, - models.InteractionStatusPending, - pendingInteractionColumns, - pendingIDExpr, - models.MessageTypeClarificationRequest, - pendingInteractionColumns, - statusExpr, - pendingActionMessageOrder(driverName, "m"), - models.MessageTypePermissionRequest, - models.InteractionStatusPending, - ) + query := renderPendingInteractionQuery(pendingInteractionQueryParts{ + scopedSessions: scopedSessions, + currentTurnOrder: orderBy, + currentTurnPredicate: predicate, + nonTerminalSessionPredicate: nonTerminalSessionPredicate("turn_row"), + pendingIDExpr: pendingIDExpr, + clarificationMessageType: string(models.MessageTypeClarificationRequest), + statusExpr: statusExpr, + pendingStatus: string(models.InteractionStatusPending), + pendingInteractionColumns: pendingInteractionColumns, + permissionOrder: pendingActionMessageOrder(driverName, "m"), + permissionMessageType: string(models.MessageTypePermissionRequest), + }) if kindConditions != "" { query = "SELECT * FROM (" + query + ") interactions WHERE " + kindConditions args = append(args, kindArgs...) @@ -89,15 +84,49 @@ func buildPendingInteractionQuery( return query, args } -// pendingInteractionQueryTemplate is the format string behind -// buildPendingInteractionQuery. current_turn duplicates the ranking in -// pendingActionsBySessionQuery on purpose: both must agree, and a shared -// builder would have to thread the scoping subquery's bind arguments through -// the projection's placeholder list. The pinning test +// pendingInteractionQueryParts names every dynamic SQL fragment in the +// pending-interaction template. Named replacement keeps a template edit from +// silently shifting a long positional fmt.Sprintf argument list. +type pendingInteractionQueryParts struct { + scopedSessions string + currentTurnOrder string + currentTurnPredicate string + nonTerminalSessionPredicate string + pendingIDExpr string + clarificationMessageType string + statusExpr string + pendingStatus string + pendingInteractionColumns string + permissionOrder string + permissionMessageType string +} + +func renderPendingInteractionQuery(parts pendingInteractionQueryParts) string { + return strings.NewReplacer( + "{{scoped_sessions}}", parts.scopedSessions, + "{{current_turn_order}}", parts.currentTurnOrder, + "{{current_turn_predicate}}", parts.currentTurnPredicate, + "{{non_terminal_session_predicate}}", parts.nonTerminalSessionPredicate, + "{{pending_id_expr}}", parts.pendingIDExpr, + "{{clarification_message_type}}", parts.clarificationMessageType, + "{{status_expr}}", parts.statusExpr, + "{{pending_status}}", parts.pendingStatus, + "{{pending_interaction_columns}}", parts.pendingInteractionColumns, + "{{permission_order}}", parts.permissionOrder, + "{{permission_message_type}}", parts.permissionMessageType, + ).Replace(pendingInteractionQueryTemplate) +} + +// pendingInteractionQueryTemplate is the named template behind +// buildPendingInteractionQuery. current_turn resolves its turn through +// currentTurnAuthority, the same predicate and ordering every other +// current-turn resolution site in this package uses, so this query and the +// compact pending-action projection (GetPendingActionsBySessionIDs) always +// agree on which turn is current. The pinning test // (TestListPendingInteractionsAgreesWithPendingActionProjection) fails if the // two ever disagree on a session. const pendingInteractionQueryTemplate = ` - WITH scoped_sessions AS (%s), + WITH scoped_sessions AS ({{scoped_sessions}}), current_turn AS ( SELECT task_session_id, turn_id FROM ( @@ -105,47 +134,47 @@ const pendingInteractionQueryTemplate = ` id AS turn_id, ROW_NUMBER() OVER ( PARTITION BY task_session_id - ORDER BY started_at DESC, created_at DESC, id DESC + ORDER BY {{current_turn_order}} ) AS rn FROM task_session_turns turn_row WHERE turn_row.task_session_id IN (SELECT id FROM scoped_sessions) - AND %s - AND %s + AND {{current_turn_predicate}} + AND {{non_terminal_session_predicate}} ) ranked WHERE rn = 1 ), pending_bundles AS ( - SELECT DISTINCT m.task_session_id, %s AS pending_id + SELECT DISTINCT m.task_session_id, {{pending_id_expr}} AS pending_id FROM task_session_messages m JOIN current_turn current ON current.task_session_id = m.task_session_id AND current.turn_id = m.turn_id - WHERE m.type = '%s' - AND COALESCE(%s, '') IN ('', '%s') + WHERE m.type = '{{clarification_message_type}}' + AND COALESCE({{status_expr}}, '') IN ('', '{{pending_status}}') ), pending_clarifications AS ( - SELECT %s + SELECT {{pending_interaction_columns}} FROM task_session_messages m JOIN current_turn current ON current.task_session_id = m.task_session_id AND current.turn_id = m.turn_id JOIN pending_bundles bundle ON bundle.task_session_id = m.task_session_id - AND bundle.pending_id = %s - WHERE m.type = '%s' + AND bundle.pending_id = {{pending_id_expr}} + WHERE m.type = '{{clarification_message_type}}' ), ranked_permissions AS ( - SELECT %s, - COALESCE(%s, '') AS permission_status, + SELECT {{pending_interaction_columns}}, + COALESCE({{status_expr}}, '') AS permission_status, ROW_NUMBER() OVER ( PARTITION BY m.task_session_id - ORDER BY m.created_at DESC, %s DESC + ORDER BY m.created_at DESC, {{permission_order}} DESC ) AS rn FROM task_session_messages m JOIN current_turn current ON current.task_session_id = m.task_session_id AND current.turn_id = m.turn_id - WHERE m.type = '%s' + WHERE m.type = '{{permission_message_type}}' ) SELECT id, task_session_id, task_id, turn_id, author_type, author_id, content, requests_input, type, metadata, created_at, updated_at @@ -154,7 +183,7 @@ const pendingInteractionQueryTemplate = ` SELECT id, task_session_id, task_id, turn_id, author_type, author_id, content, requests_input, type, metadata, created_at, updated_at FROM ranked_permissions - WHERE rn = 1 AND permission_status IN ('', '%s') + WHERE rn = 1 AND permission_status IN ('', '{{pending_status}}') ` // pendingInteractionSessionScope builds the scoped_sessions subquery. Session diff --git a/apps/backend/internal/task/repository/sqlite/pending_interactions_postgres_test.go b/apps/backend/internal/task/repository/sqlite/pending_interactions_postgres_test.go index e64c952d64..05b5a97a7e 100644 --- a/apps/backend/internal/task/repository/sqlite/pending_interactions_postgres_test.go +++ b/apps/backend/internal/task/repository/sqlite/pending_interactions_postgres_test.go @@ -105,3 +105,138 @@ func TestPostgresPendingInteractionsMatchSQLiteAuthority(t *testing.T) { } } } + +// TestPostgresListPendingInteractionsSkipsLifecycleOnlyTurn mirrors +// TestListPendingInteractionsSkipsLifecycleOnlyTurn on PostgreSQL. The +// current-turn CTE's predicate and ordering (currentTurnAuthority) are built +// from dialect.JSONExtract, which emits different SQL per driver +// (json_extract on SQLite, ::jsonb->> on Postgres), so the lifecycle_only +// exclusion needs its own coverage on this dialect rather than trusting the +// SQLite run. Skips unless KANDEV_TEST_POSTGRES_DSN is set. +// +// The clarification turn is completed (not left open) so it ties the +// open-turn-first ranking key with the lifecycle-only turn; only the +// lifecycle exclusion can then rescue it. +// TestPostgresListPendingInteractionsRanksOpenTurnAheadOfNewerCompletedTurn +// pins the open-turn-first ranking key on its own. +func TestPostgresListPendingInteractionsSkipsLifecycleOnlyTurn(t *testing.T) { + db := testutil.OpenIsolatedPostgres(t, testutil.PostgresDSNFromEnv(t)) + repo, err := NewWithDB(db, db, nil) + if err != nil { + t.Fatalf("init postgres schema: %v", err) + } + ctx := context.Background() + base := time.Date(2026, time.September, 12, 9, 0, 0, 0, time.UTC) + + seedPendingActionSession(t, repo, "task-lifecycle-pg", "session-lifecycle-pg") + questionCompletedAt := base.Add(30 * time.Second) + if err := repo.CreateTurn(ctx, &models.Turn{ + ID: "turn-question-pg", + TaskSessionID: "session-lifecycle-pg", + TaskID: "task-lifecycle-pg", + StartedAt: base, + CreatedAt: base, + CompletedAt: &questionCompletedAt, + }); err != nil { + t.Fatalf("CreateTurn(turn-question-pg): %v", err) + } + createClarificationBundleMessage(t, repo, "clar-q1-pg", "task-lifecycle-pg", "session-lifecycle-pg", "turn-question-pg", "pending-lifecycle-pg", "q1", base) + createClarificationBundleMessage(t, repo, "clar-q2-pg", "task-lifecycle-pg", "session-lifecycle-pg", "turn-question-pg", "pending-lifecycle-pg", "q2", base.Add(time.Second)) + + lifecycleCompletedAt := base.Add(time.Minute) + if err := repo.CreateTurn(ctx, &models.Turn{ + ID: "turn-lifecycle-pg", + TaskSessionID: "session-lifecycle-pg", + TaskID: "task-lifecycle-pg", + StartedAt: base.Add(time.Minute), + CreatedAt: base.Add(time.Minute), + CompletedAt: &lifecycleCompletedAt, + Metadata: map[string]interface{}{models.TurnMetaKeyLifecycleOnly: true}, + }); err != nil { + t.Fatalf("CreateTurn(turn-lifecycle-pg): %v", err) + } + createPendingActionMessage(t, repo, "lifecycle-note-pg", "task-lifecycle-pg", "session-lifecycle-pg", "turn-lifecycle-pg", + models.MessageTypeMessage, "", base.Add(time.Minute)) + + got, err := repo.ListPendingInteractions(ctx, models.PendingInteractionFilter{SessionIDs: []string{"session-lifecycle-pg"}}) + if err != nil { + t.Fatalf("ListPendingInteractions: %v", err) + } + ids := interactionMessageIDs(got) + if len(ids) != 2 { + t.Fatalf("ids = %v, want both questions of the bundle stranded behind the lifecycle-only turn", ids) + } + for _, want := range []string{"clar-q1-pg", "clar-q2-pg"} { + found := false + for _, id := range ids { + if id == want { + found = true + } + } + if !found { + t.Fatalf("ids = %v, missing %q", ids, want) + } + } + + actions, err := repo.GetPendingActionsBySessionIDs(ctx, []string{"session-lifecycle-pg"}) + if err != nil { + t.Fatalf("GetPendingActionsBySessionIDs: %v", err) + } + if _, ok := actions["session-lifecycle-pg"]; !ok { + t.Fatal("pending-action projection disagrees with the interaction list: session not reported pending") + } + + bundles, err := repo.ListUnresolvedClarificationBundles(ctx, unscopedOpts(50)) + if err != nil { + t.Fatalf("ListUnresolvedClarificationBundles: %v", err) + } + foundBundle := false + for _, bundle := range bundles.Bundles { + if bundle.SessionID == "session-lifecycle-pg" { + foundBundle = true + } + } + if !foundBundle { + t.Fatal("bundle query disagrees with the interaction list: session not reported pending") + } +} + +// TestPostgresListPendingInteractionsRanksOpenTurnAheadOfNewerCompletedTurn +// mirrors TestListPendingInteractionsRanksOpenTurnAheadOfNewerCompletedTurn +// on PostgreSQL: an open turn (completed_at IS NULL) must outrank a newer +// completed turn regardless of started_at. Skips unless +// KANDEV_TEST_POSTGRES_DSN is set. +func TestPostgresListPendingInteractionsRanksOpenTurnAheadOfNewerCompletedTurn(t *testing.T) { + db := testutil.OpenIsolatedPostgres(t, testutil.PostgresDSNFromEnv(t)) + repo, err := NewWithDB(db, db, nil) + if err != nil { + t.Fatalf("init postgres schema: %v", err) + } + ctx := context.Background() + base := time.Date(2026, time.September, 12, 9, 0, 0, 0, time.UTC) + + seedPendingActionSession(t, repo, "task-openrace-pg", "session-openrace-pg") + createPendingActionTurn(t, repo, "task-openrace-pg", "session-openrace-pg", "turn-open-older-pg", base, base) + createClarificationBundleMessage(t, repo, "clar-open-q1-pg", "task-openrace-pg", "session-openrace-pg", "turn-open-older-pg", "pending-openrace-pg", "q1", base) + + newerCompletedAt := base.Add(2 * time.Minute) + if err := repo.CreateTurn(ctx, &models.Turn{ + ID: "turn-completed-newer-pg", + TaskSessionID: "session-openrace-pg", + TaskID: "task-openrace-pg", + StartedAt: base.Add(time.Minute), + CreatedAt: base.Add(time.Minute), + CompletedAt: &newerCompletedAt, + }); err != nil { + t.Fatalf("CreateTurn(turn-completed-newer-pg): %v", err) + } + + got, err := repo.ListPendingInteractions(ctx, models.PendingInteractionFilter{SessionIDs: []string{"session-openrace-pg"}}) + if err != nil { + t.Fatalf("ListPendingInteractions: %v", err) + } + ids := interactionMessageIDs(got) + if len(ids) != 1 || ids[0] != "clar-open-q1-pg" { + t.Fatalf("ids = %v, want the still-open older turn ranked as current over the newer completed turn", ids) + } +} diff --git a/apps/backend/internal/task/repository/sqlite/pending_interactions_test.go b/apps/backend/internal/task/repository/sqlite/pending_interactions_test.go index 9ca9db2054..f0c490a3c7 100644 --- a/apps/backend/internal/task/repository/sqlite/pending_interactions_test.go +++ b/apps/backend/internal/task/repository/sqlite/pending_interactions_test.go @@ -2,12 +2,49 @@ package sqlite import ( "context" + "strings" "testing" "time" "github.com/kandev/kandev/internal/task/models" ) +func TestRenderPendingInteractionQueryReplacesNamedParts(t *testing.T) { + query := renderPendingInteractionQuery(pendingInteractionQueryParts{ + scopedSessions: "scoped-sessions", + currentTurnOrder: "current-turn-order", + currentTurnPredicate: "current-turn-predicate", + nonTerminalSessionPredicate: "non-terminal-session-predicate", + pendingIDExpr: "pending-id-expr", + clarificationMessageType: "clarification-message-type", + statusExpr: "status-expr", + pendingStatus: "pending-status", + pendingInteractionColumns: "pending-interaction-columns", + permissionOrder: "permission-order", + permissionMessageType: "permission-message-type", + }) + if strings.Contains(query, "{{") { + t.Fatalf("rendered query contains an unresolved named placeholder: %s", query) + } + for _, want := range []string{ + "scoped-sessions", + "current-turn-order", + "current-turn-predicate", + "non-terminal-session-predicate", + "pending-id-expr", + "clarification-message-type", + "status-expr", + "pending-status", + "pending-interaction-columns", + "permission-order", + "permission-message-type", + } { + if !strings.Contains(query, want) { + t.Errorf("rendered query does not contain replacement %q", want) + } + } +} + func createInteractionMessage( t *testing.T, repo *Repository, @@ -298,3 +335,136 @@ func TestListPendingInteractionsReturnsWholeBundleWhenPartiallyAnswered(t *testi } } } + +// TestListPendingInteractionsSkipsLifecycleOnlyTurn pins AC-2: a session +// whose newest turn is lifecycle_only (carrying only a non-question message) +// must still report the unanswered clarification bundle sitting on the +// previous turn, agreeing with both the pending-action projection +// (GetPendingActionsBySessionIDs) and the clarification bundle query +// (ListUnresolvedClarificationBundles). Before this fix, +// buildPendingInteractionQuery ranked current_turn by raw +// started_at/created_at/id with no lifecycle exclusion, so the newer +// lifecycle-only turn won current-turn resolution and the bundle on the +// previous turn was never joined into pending_bundles — a false negative. +// +// The clarification turn is completed (not left open) so it ties the +// open-turn-first ranking key with the lifecycle-only turn; only the +// lifecycle exclusion can then rescue it. TestListPendingInteractionsRanksOpenTurnAheadOfNewerCompletedTurn +// pins the open-turn-first ranking key on its own. +func TestListPendingInteractionsSkipsLifecycleOnlyTurn(t *testing.T) { + repo := newRepoForSessionTests(t) + ctx := context.Background() + base := time.Date(2026, time.September, 12, 9, 0, 0, 0, time.UTC) + + seedPendingActionSession(t, repo, "task-lifecycle", "session-lifecycle") + questionCompletedAt := base.Add(30 * time.Second) + if err := repo.CreateTurn(ctx, &models.Turn{ + ID: "turn-question", + TaskSessionID: "session-lifecycle", + TaskID: "task-lifecycle", + StartedAt: base, + CreatedAt: base, + CompletedAt: &questionCompletedAt, + }); err != nil { + t.Fatalf("CreateTurn(turn-question): %v", err) + } + createClarificationBundleMessage(t, repo, "clar-q1", "task-lifecycle", "session-lifecycle", "turn-question", "pending-lifecycle", "q1", base) + createClarificationBundleMessage(t, repo, "clar-q2", "task-lifecycle", "session-lifecycle", "turn-question", "pending-lifecycle", "q2", base.Add(time.Second)) + + lifecycleCompletedAt := base.Add(time.Minute) + if err := repo.CreateTurn(ctx, &models.Turn{ + ID: "turn-lifecycle", + TaskSessionID: "session-lifecycle", + TaskID: "task-lifecycle", + StartedAt: base.Add(time.Minute), + CreatedAt: base.Add(time.Minute), + CompletedAt: &lifecycleCompletedAt, + Metadata: map[string]interface{}{models.TurnMetaKeyLifecycleOnly: true}, + }); err != nil { + t.Fatalf("CreateTurn(turn-lifecycle): %v", err) + } + createPendingActionMessage(t, repo, "lifecycle-note", "task-lifecycle", "session-lifecycle", "turn-lifecycle", + models.MessageTypeMessage, "", base.Add(time.Minute)) + + got, err := repo.ListPendingInteractions(ctx, models.PendingInteractionFilter{SessionIDs: []string{"session-lifecycle"}}) + if err != nil { + t.Fatalf("ListPendingInteractions: %v", err) + } + ids := interactionMessageIDs(got) + if len(ids) != 2 { + t.Fatalf("ids = %v, want both questions of the bundle stranded behind the lifecycle-only turn", ids) + } + for _, want := range []string{"clar-q1", "clar-q2"} { + found := false + for _, id := range ids { + if id == want { + found = true + } + } + if !found { + t.Fatalf("ids = %v, missing %q", ids, want) + } + } + + actions, err := repo.GetPendingActionsBySessionIDs(ctx, []string{"session-lifecycle"}) + if err != nil { + t.Fatalf("GetPendingActionsBySessionIDs: %v", err) + } + if _, ok := actions["session-lifecycle"]; !ok { + t.Fatal("pending-action projection disagrees with the interaction list: session not reported pending") + } + + bundles, err := repo.ListUnresolvedClarificationBundles(ctx, unscopedOpts(50)) + if err != nil { + t.Fatalf("ListUnresolvedClarificationBundles: %v", err) + } + foundBundle := false + for _, bundle := range bundles.Bundles { + if bundle.SessionID == "session-lifecycle" { + foundBundle = true + } + } + if !foundBundle { + t.Fatal("bundle query disagrees with the interaction list: session not reported pending") + } +} + +// TestListPendingInteractionsRanksOpenTurnAheadOfNewerCompletedTurn pins the +// other half of currentTurnAuthority's ordering: an open turn (completed_at +// IS NULL) outranks a newer completed turn regardless of started_at. Without +// that open-turn-first key, ranking by raw started_at/created_at/id would +// resolve current_turn to the newer completed turn and strand the open +// turn's clarification bundle exactly as +// TestListPendingInteractionsSkipsLifecycleOnlyTurn strands one behind a +// lifecycle-only turn — but via the ordering axis, not the lifecycle +// predicate. +func TestListPendingInteractionsRanksOpenTurnAheadOfNewerCompletedTurn(t *testing.T) { + repo := newRepoForSessionTests(t) + ctx := context.Background() + base := time.Date(2026, time.September, 12, 9, 0, 0, 0, time.UTC) + + seedPendingActionSession(t, repo, "task-openrace", "session-openrace") + createPendingActionTurn(t, repo, "task-openrace", "session-openrace", "turn-open-older", base, base) + createClarificationBundleMessage(t, repo, "clar-open-q1", "task-openrace", "session-openrace", "turn-open-older", "pending-openrace", "q1", base) + + newerCompletedAt := base.Add(2 * time.Minute) + if err := repo.CreateTurn(ctx, &models.Turn{ + ID: "turn-completed-newer", + TaskSessionID: "session-openrace", + TaskID: "task-openrace", + StartedAt: base.Add(time.Minute), + CreatedAt: base.Add(time.Minute), + CompletedAt: &newerCompletedAt, + }); err != nil { + t.Fatalf("CreateTurn(turn-completed-newer): %v", err) + } + + got, err := repo.ListPendingInteractions(ctx, models.PendingInteractionFilter{SessionIDs: []string{"session-openrace"}}) + if err != nil { + t.Fatalf("ListPendingInteractions: %v", err) + } + ids := interactionMessageIDs(got) + if len(ids) != 1 || ids[0] != "clar-open-q1" { + t.Fatalf("ids = %v, want the still-open older turn ranked as current over the newer completed turn", ids) + } +}