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 {

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