From 2c6a9c287d849d2b86a451143140cfb26449076e Mon Sep 17 00:00:00 2001 From: nova28 Date: Thu, 13 Aug 2026 00:12:56 +0800 Subject: [PATCH] feat(backend): count accepted step-completion signals (#2526) Co-authored-by: nova28 <17953305+nova28@users.noreply.github.com> Co-authored-by: Carlos Florencio --- .github/workflows/e2e-tests.yml | 9 +- AGENTS.md | 1 + .../backend/internal/mcp/handlers/handlers.go | 93 ++++++---- .../mcp/handlers/step_complete_test.go | 171 ++++++++++++++++++ .../task/repository/sqlite/session.go | 57 ++++++ .../task/repository/sqlite/session_test.go | 41 +++++ .../workflow/signalmetrics/metrics_vars.go | 58 ++++++ .../signalmetrics/metrics_vars_test.go | 88 +++++++++ ...icit-completion-signal-for-auto-advance.md | 2 +- 9 files changed, 480 insertions(+), 40 deletions(-) create mode 100644 apps/backend/internal/workflow/signalmetrics/metrics_vars.go create mode 100644 apps/backend/internal/workflow/signalmetrics/metrics_vars_test.go diff --git a/.github/workflows/e2e-tests.yml b/.github/workflows/e2e-tests.yml index e42265fb45..9e9adf63dd 100644 --- a/.github/workflows/e2e-tests.yml +++ b/.github/workflows/e2e-tests.yml @@ -391,11 +391,12 @@ jobs: working-directory: apps run: pnpm install --frozen-lockfile - # Required for the docker pull below to succeed even if the kandev-ci - # package is private on GHCR. The `e2e` / `e2e-report` jobs above don't - # need this because GitHub's `container:` integration handles auth - # implicitly using GITHUB_TOKEN. + # The kandev-ci image is public, so fork PRs can pull it anonymously. + # Their read-only GITHUB_TOKEN can still be rejected by GHCR during an + # explicit login because it belongs to the base repository, not the + # contributor's fork. Keep authenticated pulls for trusted runs. - name: Log in to GHCR + if: github.event_name != 'pull_request' || github.event.pull_request.head.repo.full_name == github.repository uses: docker/login-action@dbcb813823bdd20940b903addbd779551569679f # v4.6.0 with: registry: ghcr.io diff --git a/AGENTS.md b/AGENTS.md index 9fc383100c..4554a451d2 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -130,6 +130,7 @@ history and remains immutable. ### Observability - In dev mode (`KANDEV_MOCK_AGENT=true` or `debug.pprofEnabled`), `/debug/vars` exposes the stdlib expvar handler. Office provider-routing metrics live under `routing_*` (route attempts, fallbacks, parked runs, provider degraded/recovered counters). The metrics are also still emitted as structured `routing.metric.*` zap logs for human debugging. +- ADR 0015's step-completion-signal telemetry lives under `workflow_*`: `workflow_step_completion_signal_received_total` (`internal/workflow/signalmetrics/`), labelled by `source` and `agent_type`, counts accepted `step_complete_kandev` signals. Its separate `workflow_step_completion_signal_fallback_used_total` counter is labelled by `agent_type` and counts manual fallback uses. The fallback button does not have a production increment site yet, so the counter remains zero until that UI ships. ### GitHub Operations Skills use `gh` CLI by default. If a `gh` command fails (not installed, not authenticated, etc.), use whatever GitHub tools are available in the environment (MCP GitHub tools, API tools, etc.) to accomplish the same operation. The goal is the same — the tool may differ. diff --git a/apps/backend/internal/mcp/handlers/handlers.go b/apps/backend/internal/mcp/handlers/handlers.go index 500a36e56e..2b479c327b 100644 --- a/apps/backend/internal/mcp/handlers/handlers.go +++ b/apps/backend/internal/mcp/handlers/handlers.go @@ -34,6 +34,7 @@ import ( workflowctrl "github.com/kandev/kandev/internal/workflow/controller" workflowmodels "github.com/kandev/kandev/internal/workflow/models" workflowsvc "github.com/kandev/kandev/internal/workflow/service" + "github.com/kandev/kandev/internal/workflow/signalmetrics" v1 "github.com/kandev/kandev/pkg/api/v1" ws "github.com/kandev/kandev/pkg/websocket" "go.uber.org/zap" @@ -83,10 +84,10 @@ type MessageCreator interface { type SessionRepository interface { UpdateTaskSessionState(ctx context.Context, sessionID string, state models.TaskSessionState, errorMessage string) error GetTaskSession(ctx context.Context, id string) (*models.TaskSession, error) - // SetSessionMetadataKey is used by handleStepComplete (ADR 0015) to - // atomically write the pending-completion bag without clobbering other - // metadata keys. - SetSessionMetadataKey(ctx context.Context, sessionID, key string, value interface{}) error + // SetSessionMetadataKeyIfAbsentOrDifferentStep atomically claims the + // pending-completion bag for a step. It preserves first-signal-wins for + // concurrent requests and replaces a stale signal from an older step. + SetSessionMetadataKeyIfAbsentOrDifferentStep(ctx context.Context, sessionID, key, stepID string, value interface{}) (bool, error) } // conditionalSessionStateUpdater is implemented by repositories that can @@ -1840,13 +1841,13 @@ func classifyAddBranchError(err error) string { // (ADR 0015). The handler: // // - Loads the session and the task to identify the current workflow step. -// - Dedupes: if a pending signal already exists for the same step, returns -// {accepted: false, reason: "already_signaled"} without overwriting. -// When the session is WAITING_FOR_INPUT, the bus event is re-published -// so a failed first-attempt publish can still drive the subscriber. -// - Otherwise writes a PendingStepCompletionSignal blob under -// TaskSession.Metadata[SessionMetaKeyPendingStepCompletion] via -// SetSessionMetadataKey (json_set — preserves other metadata keys). +// - Atomically claims the pending signal for the current step. A signal for +// an older step is replaced, but concurrent signals for the same step use +// the first successful database claim. +// - If the current step is already claimed, returns {accepted: false, +// reason: "already_signaled"}. When the session is WAITING_FOR_INPUT, the +// bus event is re-published so a failed first-attempt publish can still +// drive the subscriber. // - Publishes events.WorkflowStepCompletionSignaled so the orchestrator // subscriber can drive the on_turn_complete transition for steps with // AutoAdvanceRequiresSignal=true. Steps that don't opt in ignore the @@ -1879,29 +1880,6 @@ func (h *Handlers) handleStepComplete(ctx context.Context, msg *ws.Message) (*ws return errMsg, err } - // Idempotency: if a pending signal exists for the current step, return - // without overwriting. A stale signal for a different step (left over - // from a transition that hasn't yet cleared the bag) is treated as - // absent and overwritten — the new step's signal supersedes. - // - // Re-publish on the dedup path when the session is WAITING_FOR_INPUT. - // Without this, an agent's retry after a publish failure short-circuits - // to `already_signaled` without firing the out-of-band subscriber, - // leaving the session stuck until the user replies. Publish is - // idempotent on the subscriber side (it re-checks bag + step), so a - // double-fire when the first publish actually landed is harmless. - if existing, ok := models.LoadPendingStepSignal(session.Metadata); ok && existing.StepID == task.WorkflowStepID { - if session.State == models.TaskSessionStateWaitingForInput { - if errMsg, err := h.publishStepCompletionEvent(ctx, msg, req.TaskID, req.SessionID, task.WorkflowStepID, existing); errMsg != nil { - return errMsg, err - } - } - return ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{ - "accepted": false, - "reason": "already_signaled", - }) - } - signal := models.PendingStepCompletionSignal{ StepID: task.WorkflowStepID, Source: models.StepCompletionSourceAgent, @@ -1910,13 +1888,36 @@ func (h *Handlers) handleStepComplete(ctx context.Context, msg *ws.Message) (*ws Blockers: strings.TrimSpace(req.Blockers), SignaledAt: time.Now().UTC(), } - if err := h.sessionRepo.SetSessionMetadataKey(ctx, req.SessionID, models.SessionMetaKeyPendingStepCompletion, signal); err != nil { + stored, err := h.sessionRepo.SetSessionMetadataKeyIfAbsentOrDifferentStep( + ctx, + req.SessionID, + models.SessionMetaKeyPendingStepCompletion, + task.WorkflowStepID, + signal, + ) + if err != nil { h.logger.Error("failed to persist step-completion signal", zap.String("task_id", req.TaskID), zap.String("session_id", req.SessionID), zap.Error(err)) return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, "failed to record signal", nil) } + if !stored { + return h.handleDuplicateStepComplete(ctx, msg, req.TaskID, req.SessionID, task.WorkflowStepID, session) + } + + // Counted here, at the durable bag write, not after publishStepCompletionEvent + // below: publish is a delivery concern, and a publish failure sends the agent + // down the already_signaled dedup retry path, which never reaches this call + // again for the same signal. + // + // agent_name, not agent_id: agent_id is the store's auto-generated UUID for + // the agent row (internal/agent/settings/store/sqlite.go CreateAgent), unique + // per install and per re-creation. agent_name is the registry-facing type + // ("claude", "codex") that internal/agent/runtime/lifecycle/manager_profile.go + // keys the agent registry on. + agentType, _ := session.AgentProfileSnapshot["agent_name"].(string) + signalmetrics.RecordSignalReceived(signal.Source, agentType) if errMsg, err := h.publishStepCompletionEvent(ctx, msg, req.TaskID, req.SessionID, task.WorkflowStepID, signal); errMsg != nil { return errMsg, err @@ -1929,6 +1930,28 @@ func (h *Handlers) handleStepComplete(ctx context.Context, msg *ws.Message) (*ws }) } +func (h *Handlers) handleDuplicateStepComplete( + ctx context.Context, + msg *ws.Message, + taskID, sessionID, stepID string, + session *models.TaskSession, +) (*ws.Message, error) { + // If the initial snapshot already contained this step's signal, this is a + // retry after a possible publish failure and the event must be re-published + // while the session waits. If the initial snapshot was empty, another + // concurrent request won the claim and will publish the event itself. + existing, ok := models.LoadPendingStepSignal(session.Metadata) + if ok && existing.StepID == stepID && session.State == models.TaskSessionStateWaitingForInput { + if errMsg, err := h.publishStepCompletionEvent(ctx, msg, taskID, sessionID, stepID, existing); errMsg != nil { + return errMsg, err + } + } + return ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{ + "accepted": false, + "reason": "already_signaled", + }) +} + // resolveStepCompleteTarget loads the session + task the signal applies to // and runs the up-front validation (ownership, terminal-state guard, // workflow-step presence). Returns a populated session+task pair on success, diff --git a/apps/backend/internal/mcp/handlers/step_complete_test.go b/apps/backend/internal/mcp/handlers/step_complete_test.go index c734b9db0f..e95c114911 100644 --- a/apps/backend/internal/mcp/handlers/step_complete_test.go +++ b/apps/backend/internal/mcp/handlers/step_complete_test.go @@ -3,6 +3,10 @@ package handlers import ( "context" "encoding/json" + "expvar" + "strconv" + "sync" + "sync/atomic" "testing" "time" @@ -10,6 +14,7 @@ import ( "github.com/stretchr/testify/require" "github.com/kandev/kandev/internal/events" + "github.com/kandev/kandev/internal/events/bus" "github.com/kandev/kandev/internal/task/models" sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite" "github.com/kandev/kandev/internal/task/service" @@ -17,6 +22,47 @@ import ( ws "github.com/kandev/kandev/pkg/websocket" ) +// readSignalReceivedCounterExact reads the exact-match entry in the ADR 0015 +// workflow_step_completion_signal_received_total expvar map for the given +// key, verifying the label the handler actually wrote rather than a prefix +// that would also match a wrong or empty value. Callers seed a distinct +// agent_id per test (see seedAgentProfileSnapshot) so the exact key is also +// unique across this file's process-wide expvar state. +func readSignalReceivedCounterExact(t *testing.T, key string) int64 { + t.Helper() + v := expvar.Get("workflow_step_completion_signal_received_total") + require.NotNil(t, v, "workflow_step_completion_signal_received_total must be published") + m, ok := v.(*expvar.Map) + require.True(t, ok, "workflow_step_completion_signal_received_total must be an expvar.Map") + var total int64 + m.Do(func(kv expvar.KeyValue) { + if kv.Key != key { + return + } + n, err := strconv.ParseInt(kv.Value.String(), 10, 64) + require.NoError(t, err, "counter %q value not int: %s", kv.Key, kv.Value.String()) + total += n + }) + return total +} + +// seedAgentProfileSnapshot writes an AgentProfileSnapshot onto the session +// with distinct agent_id and agent_name values, matching the shape +// executor.resolveAgentProfileSnapshot produces in production: agent_id is +// the store's auto-generated UUID for the agent row +// (internal/agent/settings/store/sqlite.go CreateAgent), agent_name is the +// registry-facing type like "claude"/"codex" +// (internal/agent/settings/controller/reconciler.go ensureDBAgent). Seeding +// them with different values means a test asserting on agentName fails if +// the handler is ever switched back to reading agent_id. +func seedAgentProfileSnapshot(t *testing.T, repo *sqliterepo.Repository, sessionID, agentName string) { + t.Helper() + require.NoError(t, repo.UpdateTaskSessionAgentProfileSnapshot(context.Background(), sessionID, map[string]interface{}{ + "agent_id": "decoy-uuid-" + agentName, + "agent_name": agentName, + })) +} + // seedStepCompleteTarget seeds a workspace, task (with WorkflowStepID), and // session in the requested state. Used by every TestHandleStepComplete_* case // that needs the precondition chain in `resolveStepCompleteTarget` to succeed. @@ -58,6 +104,39 @@ func newStepCompleteHandler(t *testing.T, taskSvc *service.Service, repo *sqlite } } +type stepCompleteSessionReadBarrier struct { + *sqliterepo.Repository + ready chan<- struct{} + release <-chan struct{} + reads atomic.Int32 +} + +func (r *stepCompleteSessionReadBarrier) GetTaskSession(ctx context.Context, id string) (*models.TaskSession, error) { + if r.reads.Add(1) <= 2 { + r.ready <- struct{}{} + <-r.release + } + return r.Repository.GetTaskSession(ctx, id) +} + +type concurrentStepCompleteEventBus struct { + mu sync.Mutex + events []*bus.Event +} + +func (b *concurrentStepCompleteEventBus) Publish(_ context.Context, _ string, event *bus.Event) error { + b.mu.Lock() + defer b.mu.Unlock() + b.events = append(b.events, event) + return nil +} + +func (b *concurrentStepCompleteEventBus) eventCount() int { + b.mu.Lock() + defer b.mu.Unlock() + return len(b.events) +} + // TestHandleStepComplete_MissingFields covers the input-validation branches // that fail before any DB lookup. All three return ErrorCodeValidation. func TestHandleStepComplete_MissingFields(t *testing.T) { @@ -149,9 +228,13 @@ func TestHandleStepComplete_TerminalSessionRejected(t *testing.T) { func TestHandleStepComplete_FirstCallAccepted(t *testing.T) { svc, repo := newTestTaskService(t) seedStepCompleteTarget(t, repo, "task-first", "session-first", "step-1", models.TaskSessionStateRunning) + seedAgentProfileSnapshot(t, repo, "session-first", "claude-first-call") bus := &mcpRecordingEventBus{} h := newStepCompleteHandler(t, svc, repo, bus) + const counterKey = "source=agent;agent_type=claude-first-call" + before := readSignalReceivedCounterExact(t, counterKey) + msg := makeWSMessage(t, ws.ActionMCPStepComplete, map[string]interface{}{ "task_id": "task-first", "session_id": "session-first", @@ -197,6 +280,9 @@ func TestHandleStepComplete_FirstCallAccepted(t *testing.T) { assert.Equal(t, "implementation finished", data["summary"]) _, hasHandoff := data["handoff"] assert.False(t, hasHandoff, "handoff is bag-only, not on the wire") + + after := readSignalReceivedCounterExact(t, counterKey) + assert.Equal(t, int64(1), after-before, "accepted signal must increment workflow_step_completion_signal_received_total") } // TestHandleStepComplete_DedupRunningNoRepublish covers the @@ -214,9 +300,13 @@ func TestHandleStepComplete_DedupRunningNoRepublish(t *testing.T) { Summary: "first call", SignaledAt: time.Now().UTC(), })) + seedAgentProfileSnapshot(t, repo, "session-dup", "claude-dup-call") bus := &mcpRecordingEventBus{} h := newStepCompleteHandler(t, svc, repo, bus) + const counterKey = "source=agent;agent_type=claude-dup-call" + before := readSignalReceivedCounterExact(t, counterKey) + msg := makeWSMessage(t, ws.ActionMCPStepComplete, map[string]interface{}{ "task_id": "task-dup", "session_id": "session-dup", @@ -233,6 +323,9 @@ func TestHandleStepComplete_DedupRunningNoRepublish(t *testing.T) { assert.Equal(t, "already_signaled", payload["reason"]) assert.Empty(t, bus.events, "RUNNING dedup path must not re-publish (inline turn-end will consume the bag)") + + after := readSignalReceivedCounterExact(t, counterKey) + assert.Equal(t, before, after, "already_signaled dedup must not increment workflow_step_completion_signal_received_total") } // TestHandleStepComplete_DedupWaitingRepublishes covers the retry-after- @@ -272,3 +365,81 @@ func TestHandleStepComplete_DedupWaitingRepublishes(t *testing.T) { require.Len(t, bus.events, 1, "WAITING dedup must re-publish the bus event so the subscriber can drive the transition") assert.Equal(t, events.WorkflowStepCompletionSignaled, bus.events[0].Type) } + +// TestHandleStepComplete_ConcurrentCallsClaimOneSignal verifies that two +// requests which read the empty bag at the same time still produce one +// accepted signal, one event, and one telemetry increment. +func TestHandleStepComplete_ConcurrentCallsClaimOneSignal(t *testing.T) { + ctx := context.Background() + svc, repo := newTestTaskService(t) + seedStepCompleteTarget(t, repo, "task-concurrent", "session-concurrent", "step-1", models.TaskSessionStateWaitingForInput) + seedAgentProfileSnapshot(t, repo, "session-concurrent", "claude-concurrent") + + ready := make(chan struct{}, 2) + release := make(chan struct{}) + gatedRepo := &stepCompleteSessionReadBarrier{ + Repository: repo, + ready: ready, + release: release, + } + eventBus := &concurrentStepCompleteEventBus{} + h := &Handlers{ + taskSvc: svc, + sessionRepo: gatedRepo, + eventBus: eventBus, + logger: testLogger(t).WithFields(), + } + + const counterKey = "source=agent;agent_type=claude-concurrent" + before := readSignalReceivedCounterExact(t, counterKey) + messages := []*ws.Message{ + makeWSMessage(t, ws.ActionMCPStepComplete, map[string]interface{}{ + "task_id": "task-concurrent", + "session_id": "session-concurrent", + "summary": "first concurrent signal", + }), + makeWSMessage(t, ws.ActionMCPStepComplete, map[string]interface{}{ + "task_id": "task-concurrent", + "session_id": "session-concurrent", + "summary": "second concurrent signal", + }), + } + type result struct { + response *ws.Message + err error + } + results := make(chan result, len(messages)) + for _, msg := range messages { + go func(msg *ws.Message) { + response, err := h.handleStepComplete(ctx, msg) + results <- result{response: response, err: err} + }(msg) + } + <-ready + <-ready + close(release) + + accepted := 0 + for range messages { + outcome := <-results + require.NoError(t, outcome.err) + require.NotNil(t, outcome.response) + var payload map[string]interface{} + require.NoError(t, json.Unmarshal(outcome.response.Payload, &payload)) + if payload["accepted"] == true { + accepted++ + } + } + + assert.Equal(t, 1, accepted, "only one concurrent request can claim the current workflow step") + assert.Equal(t, 1, eventBus.eventCount(), "only the winning request can publish the completion event") + after := readSignalReceivedCounterExact(t, counterKey) + assert.Equal(t, int64(1), after-before, "only the winning request can increment received-signal telemetry") + + session, err := repo.GetTaskSession(ctx, "session-concurrent") + require.NoError(t, err) + signal, ok := models.LoadPendingStepSignal(session.Metadata) + require.True(t, ok, "the winning request must persist the completion signal") + assert.Equal(t, "step-1", signal.StepID) + assert.Contains(t, []string{"first concurrent signal", "second concurrent signal"}, signal.Summary) +} diff --git a/apps/backend/internal/task/repository/sqlite/session.go b/apps/backend/internal/task/repository/sqlite/session.go index 9abbbfb2e9..e500c2e2f1 100644 --- a/apps/backend/internal/task/repository/sqlite/session.go +++ b/apps/backend/internal/task/repository/sqlite/session.go @@ -1209,6 +1209,35 @@ func (r *Repository) SetSessionMetadataKeyIfAbsent( return rows > 0, nil } +// SetSessionMetadataKeyIfAbsentOrDifferentStep atomically writes a metadata +// key when it is absent or its stored value has a different step_id. The +// returned bool reports whether this call stored the value. +func (r *Repository) SetSessionMetadataKeyIfAbsentOrDifferentStep( + ctx context.Context, + sessionID, key, stepID string, + value interface{}, +) (bool, error) { + valueJSON, err := json.Marshal(value) + if err != nil { + return false, fmt.Errorf("failed to serialize metadata value: %w", err) + } + now := time.Now().UTC() + driver := r.db.DriverName() + path := key + stepPath := key + if !dialect.IsPostgres(driver) { + path = "$." + key + stepPath = path + ".step_id" + } + result, err := r.db.ExecContext(ctx, r.db.Rebind(setSessionMetadataKeyIfAbsentOrDifferentStepQuery(driver)), + path, string(valueJSON), now, sessionID, stepPath, stepID) + if err != nil { + return false, err + } + rows, _ := result.RowsAffected() + return rows > 0, nil +} + // SetSessionMetadataKeyIfAbsentIfState atomically claims a metadata key only // while the session remains in expectedState. It is used when a terminal // transition owns a one-time side effect that must not be emitted by a stale @@ -1286,6 +1315,34 @@ func setSessionMetadataKeyIfAbsentQuery(driver string) string { ` } +func setSessionMetadataKeyIfAbsentOrDifferentStepQuery(driver string) string { + if dialect.IsPostgres(driver) { + base := "CASE WHEN metadata IS NULL OR metadata = 'null' OR metadata = '' THEN '{}'::jsonb ELSE metadata::jsonb END" + return ` + UPDATE task_sessions + SET metadata = jsonb_set( + ` + base + `, + ARRAY[?]::text[], + ?::jsonb, + true + )::text, + updated_at = ? + WHERE id = ? + AND jsonb_extract_path_text(` + base + `, ?, 'step_id') IS DISTINCT FROM ? + ` + } + base := "CASE WHEN metadata IS NULL OR metadata = 'null' OR metadata = '' THEN '{}' ELSE metadata END" + return ` + UPDATE task_sessions + SET metadata = json_set(` + base + `, ?, json(?)), + updated_at = ? + WHERE id = ? + AND ( + json_extract(` + base + `, ?) IS NOT ? + ) + ` +} + func setSessionMetadataKeyIfAbsentIfStateQuery(driver string) string { return setSessionMetadataKeyIfAbsentQuery(driver) + " AND state = ?" } diff --git a/apps/backend/internal/task/repository/sqlite/session_test.go b/apps/backend/internal/task/repository/sqlite/session_test.go index 70bff0a4bd..79ada6c6d2 100644 --- a/apps/backend/internal/task/repository/sqlite/session_test.go +++ b/apps/backend/internal/task/repository/sqlite/session_test.go @@ -645,6 +645,37 @@ func TestSetSessionMetadataKeyIfAbsentSQLiteIsWriteOnce(t *testing.T) { } } +func TestSetSessionMetadataKeyIfAbsentOrDifferentStepSQLiteReplacesOnlyStaleStep(t *testing.T) { + repo := newRepoForSessionTests(t) + seedForMsgTest(t, repo, "task-step-claim", "session-step-claim", "turn-step-claim") + ctx := context.Background() + + first := models.PendingStepCompletionSignal{StepID: "step-1", Summary: "first"} + stored, err := repo.SetSessionMetadataKeyIfAbsentOrDifferentStep( + ctx, "session-step-claim", models.SessionMetaKeyPendingStepCompletion, "step-1", first) + require.NoError(t, err) + require.True(t, stored, "an empty signal bag should be claimed") + + second := models.PendingStepCompletionSignal{StepID: "step-2", Summary: "second"} + stored, err = repo.SetSessionMetadataKeyIfAbsentOrDifferentStep( + ctx, "session-step-claim", models.SessionMetaKeyPendingStepCompletion, "step-2", second) + require.NoError(t, err) + require.True(t, stored, "a signal from an older step should be replaced") + + third := models.PendingStepCompletionSignal{StepID: "step-2", Summary: "third"} + stored, err = repo.SetSessionMetadataKeyIfAbsentOrDifferentStep( + ctx, "session-step-claim", models.SessionMetaKeyPendingStepCompletion, "step-2", third) + require.NoError(t, err) + require.False(t, stored, "a signal for the current step should keep the first payload") + + session, err := repo.GetTaskSession(ctx, "session-step-claim") + require.NoError(t, err) + signal, ok := models.LoadPendingStepSignal(session.Metadata) + require.True(t, ok) + require.Equal(t, "step-2", signal.StepID) + require.Equal(t, "second", signal.Summary) +} + func TestUpdateSessionContextWindowSQLiteCountsStrictUsageDrops(t *testing.T) { repo := newRepoForSessionTests(t) ctx := context.Background() @@ -700,6 +731,16 @@ func TestSetSessionMetadataKeyIfAbsentQueryUsesPostgresJSONB(t *testing.T) { } } +func TestSetSessionMetadataKeyIfAbsentOrDifferentStepQueryUsesPostgresJSONB(t *testing.T) { + query := setSessionMetadataKeyIfAbsentOrDifferentStepQuery(dialect.PGX) + if strings.Contains(query, "json_set") || strings.Contains(query, "json_extract") || strings.Contains(query, "json(?)") { + t.Fatalf("postgres step-aware claim query uses SQLite JSON functions: %s", query) + } + if !strings.Contains(query, "jsonb_set") || !strings.Contains(query, "jsonb_extract_path_text") || !strings.Contains(query, "IS DISTINCT FROM") { + t.Fatalf("postgres step-aware claim query must use atomic JSONB comparison: %s", query) + } +} + func TestListTaskSessionWorktreesFiltersInactiveRows(t *testing.T) { repo := newRepoForSessionTests(t) ctx := context.Background() diff --git a/apps/backend/internal/workflow/signalmetrics/metrics_vars.go b/apps/backend/internal/workflow/signalmetrics/metrics_vars.go new file mode 100644 index 0000000000..9fa2592655 --- /dev/null +++ b/apps/backend/internal/workflow/signalmetrics/metrics_vars.go @@ -0,0 +1,58 @@ +// Package signalmetrics publishes the ADR 0015 step-completion-signal expvar +// counters. It is a neutral package so both MCP and UI code can record the +// signal source without depending on each other. +package signalmetrics + +import ( + "expvar" + "strings" +) + +// workflowStepSignalReceived counts accepted `step_complete_kandev` signals +// (ADR 0015), labelled by source and agent type. It is the ADR's +// `step_completion_signal_received_total` counter. +var workflowStepSignalReceived = expvar.NewMap("workflow_step_completion_signal_received_total") + +// workflowStepSignalFallbackUsed counts manual completion fallbacks, labelled +// by agent type. It is separate from workflowStepSignalReceived so the ADR's +// fallback-used / received ratio has the correct denominator. +var workflowStepSignalFallbackUsed = expvar.NewMap("workflow_step_completion_signal_fallback_used_total") + +// metricLabel builds a "k1=v1;k2=v2;..." label string for an expvar map key, +// matching the convention in internal/office/scheduler/metrics_vars.go so a +// downstream Prometheus translation layer can split on the same delimiters. +// Empty values are still emitted so a downstream parser sees consistent +// cardinality dimensions. Keys are intentionally NOT escaped — callers +// control the inputs (source constants, agent profile IDs). +func metricLabel(pairs ...string) string { + if len(pairs)%2 != 0 { + return "" + } + parts := make([]string, 0, len(pairs)/2) + for i := 0; i < len(pairs); i += 2 { + parts = append(parts, pairs[i]+"="+pairs[i+1]) + } + return strings.Join(parts, ";") +} + +// RecordSignalReceived counts one accepted step-completion signal. `source` +// is one of models.StepCompletionSource*. The manual fallback affordance may +// call RecordSignalFallbackUsed separately when it is available. +// `agentType` is the bounded agent-type dimension the ADR's ratio threshold +// ("fallback-used / received <= 10% per agent type") is expressed against — +// e.g. AgentProfile.AgentName ("claude", "codex"), the registry-facing type +// (internal/agent/runtime/lifecycle/manager_profile.go keys the agent +// registry on it) — not AgentProfile.AgentID, which is the store's +// auto-generated UUID for the agent row +// (internal/agent/settings/store/sqlite.go CreateAgent) and is therefore +// unique per install and per agent re-creation, not a bounded type. +func RecordSignalReceived(source, agentType string) { + workflowStepSignalReceived.Add( + metricLabel("source", source, "agent_type", agentType), 1) +} + +// RecordSignalFallbackUsed counts one manual completion fallback. The +// separate counter keeps fallback events out of the received-signal count. +func RecordSignalFallbackUsed(agentType string) { + workflowStepSignalFallbackUsed.Add(metricLabel("agent_type", agentType), 1) +} diff --git a/apps/backend/internal/workflow/signalmetrics/metrics_vars_test.go b/apps/backend/internal/workflow/signalmetrics/metrics_vars_test.go new file mode 100644 index 0000000000..cef7977b18 --- /dev/null +++ b/apps/backend/internal/workflow/signalmetrics/metrics_vars_test.go @@ -0,0 +1,88 @@ +package signalmetrics + +import ( + "expvar" + "strconv" + "strings" + "testing" +) + +// readCounter walks the expvar map looking for a key that matches the +// supplied prefix. Returns 0 when no key matches. The prefix match keeps the +// assertion robust against process-wide test pollution (other tests in this +// package may push entries with different labels). +func readCounter(t *testing.T, m *expvar.Map, prefix string) int64 { + t.Helper() + var total int64 + m.Do(func(kv expvar.KeyValue) { + if !strings.HasPrefix(kv.Key, prefix) { + return + } + n, err := strconv.ParseInt(kv.Value.String(), 10, 64) + if err != nil { + t.Fatalf("counter %q value not int: %s", kv.Key, kv.Value.String()) + } + total += n + }) + return total +} + +func TestMetricLabel(t *testing.T) { + cases := []struct { + name string + pairs []string + want string + }{ + {"single_pair", []string{"source", "agent"}, "source=agent"}, + {"empty_agent_type", []string{"source", "agent", "agent_type", ""}, + "source=agent;agent_type="}, + {"odd_args_returns_empty", []string{"source"}, ""}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + if got := metricLabel(tc.pairs...); got != tc.want { + t.Errorf("metricLabel(%v) = %q, want %q", tc.pairs, got, tc.want) + } + }) + } +} + +func TestRecordSignalReceived(t *testing.T) { + label := metricLabel("source", "agent", "agent_type", "claude-metrics-test") + + before := readCounter(t, workflowStepSignalReceived, label) + RecordSignalReceived("agent", "claude-metrics-test") + after := readCounter(t, workflowStepSignalReceived, label) + + if after-before != 1 { + t.Errorf("received counter delta = %d, want 1", after-before) + } +} + +func TestRecordSignalFallbackUsed(t *testing.T) { + const agentType = "claude-fallback-test" + label := metricLabel("agent_type", agentType) + receivedLabel := metricLabel("source", "manual_fallback", "agent_type", agentType) + + before := readCounter(t, workflowStepSignalFallbackUsed, label) + receivedBefore := readCounter(t, workflowStepSignalReceived, receivedLabel) + RecordSignalFallbackUsed(agentType) + after := readCounter(t, workflowStepSignalFallbackUsed, label) + receivedAfter := readCounter(t, workflowStepSignalReceived, receivedLabel) + + if after-before != 1 { + t.Errorf("fallback counter delta = %d, want 1", after-before) + } + if receivedAfter != receivedBefore { + t.Errorf("received counter changed for fallback-only record: before=%d after=%d", receivedBefore, receivedAfter) + } +} + +func TestWorkflowStepSignalReceivedPublishedAtKnownName(t *testing.T) { + if expvar.Get("workflow_step_completion_signal_received_total") == nil { + t.Error("expvar \"workflow_step_completion_signal_received_total\" not published — /debug/vars consumers will miss it") + } + if expvar.Get("workflow_step_completion_signal_fallback_used_total") == nil { + t.Error("expvar \"workflow_step_completion_signal_fallback_used_total\" not published — /debug/vars consumers will miss it") + } +} diff --git a/docs/decisions/0015-explicit-completion-signal-for-auto-advance.md b/docs/decisions/0015-explicit-completion-signal-for-auto-advance.md index de4d03aa6b..2eb744c3d9 100644 --- a/docs/decisions/0015-explicit-completion-signal-for-auto-advance.md +++ b/docs/decisions/0015-explicit-completion-signal-for-auto-advance.md @@ -93,7 +93,7 @@ Today, a workflow step's `on_turn_complete` actions (most commonly `move_to_next - **Clarification timeout becomes a hard pause, not a completed turn.** A no-answer `ask_user_question_kandev` result cancels the current agent turn without running `on_turn_complete`. The user-visible state is a waiting session with the clarification still answerable; the next user answer resumes the same step instead of triggering turn-start automation. - **Idempotency.** Multiple calls within the same step are deduped at the bag-write site (first call wins, subsequent calls return `{accepted: false, reason: "already_signaled"}` without error). Cross-step re-entry is allowed — a fresh step's `currentStepID` no longer matches the stale bag entry, so the next call writes through. - **No new schema for the signal itself.** Storage piggybacks on the existing `task_sessions.metadata` JSON column. Audit of *consumed* signals piggybacks on the existing `session_step_history.metadata` JSON column. Cancelled signals (overwritten by a user reply) are not retained — chat already surfaces what the user needs. -- **Telemetry (deferred).** The plan is two `expvar` counters under `workflow_*` — `step_completion_signal_received_total` and `step_completion_signal_fallback_used_total` — whose ratio (fallback-used / received ≤ 10% per agent type) becomes the headline "is the explicit-signal flow working" metric. Neither counter exists yet; ship in a follow-up alongside the manual fallback button so both halves of the ratio can be measured at once. +- **Telemetry.** Two `expvar` counters under `workflow_*` expose the explicit-signal health metric: `step_completion_signal_received_total` counts accepted signals by source and agent type, and `step_completion_signal_fallback_used_total` counts manual fallback uses by agent type. Their ratio (fallback-used / received ≤ 10% per agent type) is the headline "is the explicit-signal flow working" metric. The fallback counter is published now, but its increment site waits for the manual fallback button. - **No data-model coupling to `task_repositories` or `task_prs`.** The signal is per-session, per-step; it does not interact with the multi-branch work in ADR 0013. ## Alternatives considered