diff --git a/adapters/postgres/continuity_repository.go b/adapters/postgres/continuity_repository.go index 8989c26..1775c35 100644 --- a/adapters/postgres/continuity_repository.go +++ b/adapters/postgres/continuity_repository.go @@ -234,6 +234,35 @@ func (r *ContinuityRepository) ResponseCandidates(ctx context.Context, owner ide if owner == "" || limit < 1 || limit > 50 { return nil, fmt.Errorf("invalid response candidate query") } + candidates := make([]continuitydomain.ResponseCandidate, 0, limit) + attentionRows, err := r.pool.Query(ctx, `SELECT + attention_item_id,owner_principal_id,scope_id,source_kind,source_prospective_item_id,source_external_ref,level,state,summary, + due_at_unix_millis,created_at_unix_millis,delivered_at_unix_millis,action_intent_id,authorized_action_id,action_attempt_id,delivery_outcome, + response_id,response_utterance_id,response_operation,response_outcome,responded_at_unix_millis,revision + FROM continuity.attention_items + WHERE owner_principal_id=$1 + AND source_kind=$2 + AND state IN ($3,$4,$5) + ORDER BY created_at_unix_millis DESC, attention_item_id DESC + LIMIT $6`, + owner, continuity.AttentionSourceAgentDelegation, continuity.AttentionPending, continuity.AttentionDelivered, continuity.AttentionFailed, limit) + if err != nil { + return nil, fmt.Errorf("list delegation response candidates: %w", err) + } + defer attentionRows.Close() + for attentionRows.Next() { + item, err := scanAttentionItem(attentionRows) + if err != nil { + return nil, fmt.Errorf("scan delegation response candidate: %w", err) + } + candidates = append(candidates, continuitydomain.ResponseCandidate{Attention: item}) + } + if err := attentionRows.Err(); err != nil { + return nil, fmt.Errorf("list delegation response candidates: %w", err) + } + if len(candidates) >= limit { + return candidates[:limit], nil + } rows, err := r.pool.Query(ctx, `SELECT COALESCE(a.attention_item_id,''),COALESCE(a.owner_principal_id,''),COALESCE(a.scope_id,''),COALESCE(a.source_kind,''),COALESCE(a.source_prospective_item_id,''),COALESCE(a.source_external_ref,''),COALESCE(a.level,''),COALESCE(a.state,''),COALESCE(a.summary,''), COALESCE(a.due_at_unix_millis,0),COALESCE(a.created_at_unix_millis,0),COALESCE(a.delivered_at_unix_millis,0),COALESCE(a.action_intent_id,''),COALESCE(a.authorized_action_id,''),COALESCE(a.action_attempt_id,''),COALESCE(a.delivery_outcome,''), @@ -247,12 +276,11 @@ func (r *ContinuityRepository) ResponseCandidates(ctx context.Context, owner ide ORDER BY p.recorded_at DESC, COALESCE(a.delivered_at_unix_millis,0) DESC, p.prospective_item_id DESC LIMIT $8`, owner, continuity.AttentionSourceProspectiveReminder, continuity.AttentionDelivered, continuity.AttentionPending, - continuity.AttentionFailed, continuity.ProspectiveDelivered, continuity.ProspectiveArmed, limit) + continuity.AttentionFailed, continuity.ProspectiveDelivered, continuity.ProspectiveArmed, limit-len(candidates)) if err != nil { return nil, fmt.Errorf("list human response candidates: %w", err) } defer rows.Close() - var candidates []continuitydomain.ResponseCandidate for rows.Next() { var candidate continuitydomain.ResponseCandidate var attentionID, attentionOwner, attentionScope, sourceKind, sourceProspective, sourceExternalRef, level, attentionState, attentionSummary string @@ -303,6 +331,47 @@ func (r *ContinuityRepository) ResponseCandidates(ctx context.Context, owner ide return candidates, nil } +func scanAttentionItem(scanner interface { + Scan(...any) error +}) (continuity.AttentionItem, error) { + var item continuity.AttentionItem + var attentionID, attentionOwner, attentionScope, sourceKind, sourceProspective, sourceExternalRef, level, attentionState, attentionSummary string + var actionIntent, authorizedAction, actionAttempt, deliveryOutcome, responseID, responseUtterance, responseOperation, responseOutcome string + var dueAt, createdAt, deliveredAt, respondedAt int64 + var attentionRevision uint64 + if err := scanner.Scan( + &attentionID, &attentionOwner, &attentionScope, &sourceKind, &sourceProspective, &sourceExternalRef, &level, &attentionState, &attentionSummary, + &dueAt, &createdAt, &deliveredAt, &actionIntent, &authorizedAction, &actionAttempt, &deliveryOutcome, + &responseID, &responseUtterance, &responseOperation, &responseOutcome, &respondedAt, &attentionRevision); err != nil { + return continuity.AttentionItem{}, err + } + item = continuity.AttentionItem{ + ID: continuity.AttentionItemID(attentionID), + OwnerPrincipalID: identity.PrincipalID(attentionOwner), + ScopeID: authority.ScopeID(attentionScope), + SourceKind: continuity.AttentionSourceKind(sourceKind), + SourceProspectiveID: continuity.ProspectiveItemID(sourceProspective), + SourceExternalRef: sourceExternalRef, + Level: continuity.AttentionLevel(level), + State: continuity.AttentionState(attentionState), + Summary: attentionSummary, + DueAtUnixMillis: dueAt, + CreatedAtUnixMillis: createdAt, + DeliveredAtUnixMillis: deliveredAt, + ActionIntentID: continuity.CommittedActionIntentID(actionIntent), + AuthorizedActionID: continuity.CommittedAuthorizedActionID(authorizedAction), + ActionAttemptID: continuity.CommittedActionAttemptID(actionAttempt), + DeliveryOutcome: deliveryOutcome, + ResponseID: continuity.HumanResponseID(responseID), + ResponseUtteranceID: responseUtterance, + ResponseOperation: continuity.HumanResponseOperation(responseOperation), + ResponseOutcome: continuity.ResponseOutcome(responseOutcome), + RespondedAtUnixMillis: respondedAt, + Revision: continuity.AttentionRevision(attentionRevision), + } + return item, nil +} + func (r *ContinuityRepository) ApplyHumanResponse(ctx context.Context, response continuity.HumanResponse) error { tx, err := r.pool.Begin(ctx) if err != nil { diff --git a/cmd/continuity-runtime/human_response_test.go b/cmd/continuity-runtime/human_response_test.go index da98e85..46a73c9 100644 --- a/cmd/continuity-runtime/human_response_test.go +++ b/cmd/continuity-runtime/human_response_test.go @@ -65,6 +65,39 @@ func TestCreatorResponseEndpointRecordsAmbiguousWithoutSilentTargetMutation(t *t } } +func TestCreatorResponseEndpointContinuesDelegationAttentionViaNaturalMessage(t *testing.T) { + core := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _ = json.NewEncoder(w).Encode(identity.CreatorSession{ActorID: "actor-1", PrincipalID: "principal-1"}) + })) + defer core.Close() + identityClient, err := newCreatorIdentityClient(core.URL, core.Client()) + if err != nil { + t.Fatal(err) + } + repository := &fakeResponseRepository{candidates: []continuitydomain.ResponseCandidate{ + {Attention: continuity.AttentionItem{ + ID: "work:delegation:delegation-1", + OwnerPrincipalID: "principal-1", + ScopeID: "scope-1", + SourceKind: continuity.AttentionSourceAgentDelegation, + SourceExternalRef: "work:delegation:delegation-1", + State: continuity.AttentionPending, + Summary: "Нужен твой выбор по задаче coding-agent: docs/PROJECT_STATUS.md", + Revision: 1, + }}, + {Attention: continuity.AttentionItem{ID: "attention-1", SourceProspectiveID: "prospective-1", State: continuity.AttentionDelivered, Summary: "старое напоминание"}}, + }} + mux := http.NewServeMux() + registerCreatorResponseRoute(mux, identityClient, repository, func() time.Time { return time.Unix(100, 0) }) + request := httptest.NewRequest(http.MethodPost, "/v1/creator/responses", strings.NewReader(`{"SourceUtteranceID":"utterance-1","SourceIntentID":"intent-1","Message":"Продолжай задачу coding-agent по docs/PROJECT_STATUS.md","RequestedAtUnixMillis":100000}`)) + request.Header.Set("Authorization", "Bearer ok") + recorder := httptest.NewRecorder() + mux.ServeHTTP(recorder, request) + if recorder.Code != http.StatusCreated || repository.applied.Operation != continuity.ResponseContinue || repository.applied.TargetExternalRef != "work:delegation:delegation-1" || repository.applied.Outcome != continuity.ResponseOutcomeContinued { + t.Fatalf("status=%d applied=%#v body=%s", recorder.Code, repository.applied, recorder.Body.String()) + } +} + type fakeResponseRepository struct { candidates []continuitydomain.ResponseCandidate applied continuity.HumanResponse