fix(task): stop plugin pending-interaction API from picking the wrong current turn (#3618)

Co-authored-by: nova28 <17953305+nova28@users.noreply.github.com>
Co-authored-by: Carlos Florencio <carlosmflorencio@gmail.com>
This commit is contained in:
Henry
2026-09-13 06:01:01 +08:00
committed by GitHub
parent aebb01e714
commit 2dc1596667
3 changed files with 373 additions and 39 deletions
@@ -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
@@ -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, "<missing>", 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)
}
}
@@ -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, "<missing>", 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)
}
}