From 18fbc7f105e3d2445e1ca94af9a71e3d68234fd6 Mon Sep 17 00:00:00 2001 From: Mikhail Date: Mon, 17 Aug 2026 16:33:05 +0300 Subject: [PATCH] DEV-000065 fix notification binding origin --- adapters/postgres/continuity_repository.go | 19 ++++++++ cmd/continuity-runtime/due_worker.go | 12 ++++++ cmd/continuity-runtime/due_worker_test.go | 33 ++++++++++++++ database/bootstrap/001_roles_and_schemas.sql | 14 ++++++ ..._notification_binding_assistant_origin.sql | 43 +++++++++++++++++++ docs/PROJECT_STATUS.md | 2 +- modules/continuity/ports/repository.go | 1 + 7 files changed, 123 insertions(+), 1 deletion(-) create mode 100644 database/migrations/021_notification_binding_assistant_origin.sql diff --git a/adapters/postgres/continuity_repository.go b/adapters/postgres/continuity_repository.go index 60c8cde..df53232 100644 --- a/adapters/postgres/continuity_repository.go +++ b/adapters/postgres/continuity_repository.go @@ -228,6 +228,25 @@ func (r *ContinuityRepository) MarkAttentionDelivered(ctx context.Context, id co return item, err } +func (r *ContinuityRepository) MarkAttentionFailed(ctx context.Context, id continuity.AttentionItemID, revision continuity.AttentionRevision, intentID continuity.CommittedActionIntentID, authorizedID continuity.CommittedAuthorizedActionID, attemptID continuity.CommittedActionAttemptID, outcome string) (continuity.AttentionItem, error) { + var item continuity.AttentionItem + err := r.pool.QueryRow(ctx, `UPDATE continuity.attention_items + SET state=$1, revision=revision+1, action_intent_id=$2, authorized_action_id=$3, + action_attempt_id=$4, delivery_outcome=$5 + WHERE attention_item_id=$6 AND state=$7 AND revision=$8 + RETURNING attention_item_id,owner_principal_id,scope_id,source_kind,source_prospective_item_id,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,revision`, + continuity.AttentionFailed, intentID, authorizedID, attemptID, outcome, + id, continuity.AttentionDelivering, revision).Scan( + &item.ID, &item.OwnerPrincipalID, &item.ScopeID, &item.SourceKind, &item.SourceProspectiveID, &item.Level, &item.State, + &item.Summary, &item.DueAtUnixMillis, &item.CreatedAtUnixMillis, &item.DeliveredAtUnixMillis, &item.ActionIntentID, + &item.AuthorizedActionID, &item.ActionAttemptID, &item.DeliveryOutcome, &item.Revision) + if errors.Is(err, pgx.ErrNoRows) { + return continuity.AttentionItem{}, ErrAttentionRevisionConflict + } + return item, err +} + // SatisfyDelivered owns both the Prospective CAS and its idempotency receipt in // one Continuity-local transaction. It never reaches Commit storage. func (r *ContinuityRepository) SatisfyDelivered(ctx context.Context, command continuity.SatisfyDeliveredCommand) (continuity.ActionReceipt, error) { diff --git a/cmd/continuity-runtime/due_worker.go b/cmd/continuity-runtime/due_worker.go index 6bff266..281a40d 100644 --- a/cmd/continuity-runtime/due_worker.go +++ b/cmd/continuity-runtime/due_worker.go @@ -21,6 +21,7 @@ type dueReminderRepository interface { PendingAttention(context.Context, int) ([]continuity.AttentionItem, error) MarkAttentionDelivering(context.Context, continuity.AttentionItemID, continuity.AttentionRevision) (continuity.AttentionItem, error) MarkAttentionDelivered(context.Context, continuity.AttentionItemID, continuity.AttentionRevision, continuity.CommittedActionIntentID, continuity.CommittedAuthorizedActionID, continuity.CommittedActionAttemptID, string, int64) (continuity.AttentionItem, error) + MarkAttentionFailed(context.Context, continuity.AttentionItemID, continuity.AttentionRevision, continuity.CommittedActionIntentID, continuity.CommittedAuthorizedActionID, continuity.CommittedActionAttemptID, string) (continuity.AttentionItem, error) Load(context.Context, continuity.ProspectiveItemID) (continuity.ProspectiveItem, error) } @@ -170,6 +171,17 @@ func (w dueReminderWorker) deliverPending(ctx context.Context, now time.Time) er log.Printf("attention %s notification unavailable: %v", attention.ID, err) continue } + if result.State != commitcontract.ActionSucceeded { + _, err = w.repository.MarkAttentionFailed(ctx, delivering.ID, delivering.Revision, + continuity.CommittedActionIntentID(result.ActionIntentID), + continuity.CommittedAuthorizedActionID(result.AuthorizedActionID), + continuity.CommittedActionAttemptID(result.ActionAttemptID), + string(result.State)) + if err != nil { + log.Printf("attention %s delivery failure outcome not persisted: %v", attention.ID, err) + } + continue + } _, err = w.repository.MarkAttentionDelivered(ctx, delivering.ID, delivering.Revision, continuity.CommittedActionIntentID(result.ActionIntentID), continuity.CommittedAuthorizedActionID(result.AuthorizedActionID), diff --git a/cmd/continuity-runtime/due_worker_test.go b/cmd/continuity-runtime/due_worker_test.go index 0d1cec6..f1630fe 100644 --- a/cmd/continuity-runtime/due_worker_test.go +++ b/cmd/continuity-runtime/due_worker_test.go @@ -62,6 +62,18 @@ func (r *memoryDueRepo) MarkAttentionDelivered(_ context.Context, id continuity. return r.attention, nil } +func (r *memoryDueRepo) MarkAttentionFailed(_ context.Context, id continuity.AttentionItemID, revision continuity.AttentionRevision, intentID continuity.CommittedActionIntentID, authorizedID continuity.CommittedAuthorizedActionID, attemptID continuity.CommittedActionAttemptID, outcome string) (continuity.AttentionItem, error) { + if r.attention.ID == id && r.attention.Revision == revision { + r.attention.State = continuity.AttentionFailed + r.attention.Revision++ + r.attention.ActionIntentID = intentID + r.attention.AuthorizedActionID = authorizedID + r.attention.ActionAttemptID = attemptID + r.attention.DeliveryOutcome = outcome + } + return r.attention, nil +} + func (r *memoryDueRepo) Load(context.Context, continuity.ProspectiveItemID) (continuity.ProspectiveItem, error) { return r.item, nil } @@ -73,6 +85,13 @@ func (n *fakeDueNotifier) NotifyDue(context.Context, continuity.ProspectiveItem) return dueNotificationResponse{ActionIntentID: "intent", AuthorizedActionID: "authorized", ActionAttemptID: "attempt", State: commitcontract.ActionSucceeded}, nil } +type unknownDueNotifier struct{ calls int } + +func (n *unknownDueNotifier) NotifyDue(context.Context, continuity.ProspectiveItem) (dueNotificationResponse, error) { + n.calls++ + return dueNotificationResponse{ActionIntentID: "intent", AuthorizedActionID: "authorized", ActionAttemptID: "attempt", State: commitcontract.ActionUnknownOutcome}, nil +} + func TestDueReminderProducesAttentionAndNotificationOnce(t *testing.T) { repo := &memoryDueRepo{item: continuity.ProspectiveItem{ID: "prospective", OwnerPrincipalID: identity.PrincipalID("owner"), ScopeID: authority.ScopeID("scope"), Summary: "проверить уведомление", TriggerAtUnixMillis: 10, State: continuity.ProspectiveArmed, Revision: 1}} notifier := &fakeDueNotifier{} @@ -90,3 +109,17 @@ func TestDueReminderProducesAttentionAndNotificationOnce(t *testing.T) { t.Fatal("delivery evidence was not persisted") } } + +func TestDueReminderUnknownOutcomeDoesNotMarkAttentionDelivered(t *testing.T) { + repo := &memoryDueRepo{item: continuity.ProspectiveItem{ID: "prospective", OwnerPrincipalID: identity.PrincipalID("owner"), ScopeID: authority.ScopeID("scope"), Summary: "проверить уведомление", TriggerAtUnixMillis: 10, State: continuity.ProspectiveArmed, Revision: 1}} + notifier := &unknownDueNotifier{} + now := time.UnixMilli(20) + worker := dueReminderWorker{repository: repo, commit: notifier, now: func() time.Time { return now }, newID: func() (ids.ID, error) { return ids.ID("attention"), nil }, limit: 10} + worker.process(context.Background()) + if repo.attention.State != continuity.AttentionFailed { + t.Fatalf("unknown outcome must not be delivered, got %s", repo.attention.State) + } + if repo.attention.DeliveryOutcome != string(commitcontract.ActionUnknownOutcome) { + t.Fatalf("missing unknown outcome evidence: %#v", repo.attention) + } +} diff --git a/database/bootstrap/001_roles_and_schemas.sql b/database/bootstrap/001_roles_and_schemas.sql index 82fb43d..44c42fe 100644 --- a/database/bootstrap/001_roles_and_schemas.sql +++ b/database/bootstrap/001_roles_and_schemas.sql @@ -532,6 +532,20 @@ ON CONFLICT DO NOTHING; INSERT INTO capability.capability_implementation_heads(implementation_id, revision, capability_kind, owner_principal_id) VALUES ('internal-reminder-scheduler-v1', 1, 'SCHEDULE_REMINDER', 'domovoy-creator-principal') ON CONFLICT (implementation_id) DO NOTHING; +INSERT INTO capability.notification_bindings ( + binding_id, capability_id, owner_principal_id, revision, state, https_origin, + delivery_path, receipt_path, protocol, secret_ref, verified_at_unix_millis, health +) +VALUES ( + 'domovoy-notification-binding', 'prospective-notification-delivery', 'domovoy-creator-principal', + 2, 'ACTIVE', 'https://assistant.cin.su', '/delivery/v1/deliveries', '/delivery/v1/receipts', + 'IDEMPOTENT_RECEIPT_V1', 'domovoy-notification-write-v1', 1, 'AVAILABLE' +) +ON CONFLICT DO NOTHING; +INSERT INTO capability.notification_binding_heads(capability_id, owner_principal_id, binding_id, revision) +VALUES ('prospective-notification-delivery', 'domovoy-creator-principal', 'domovoy-notification-binding', 2) +ON CONFLICT (capability_id, owner_principal_id) DO UPDATE SET binding_id=EXCLUDED.binding_id, revision=EXCLUDED.revision +WHERE capability.notification_binding_heads.revision < EXCLUDED.revision; RESET ROLE; SET ROLE continuity_role; diff --git a/database/migrations/021_notification_binding_assistant_origin.sql b/database/migrations/021_notification_binding_assistant_origin.sql new file mode 100644 index 0000000..e56aa41 --- /dev/null +++ b/database/migrations/021_notification_binding_assistant_origin.sql @@ -0,0 +1,43 @@ +BEGIN; + +SET ROLE capability_role; + +INSERT INTO capability.notification_bindings ( + binding_id, capability_id, owner_principal_id, revision, state, https_origin, + delivery_path, receipt_path, protocol, secret_ref, verified_at_unix_millis, health +) +VALUES ( + 'domovoy-notification-binding', + 'prospective-notification-delivery', + 'domovoy-creator-principal', + 2, + 'ACTIVE', + 'https://assistant.cin.su', + '/delivery/v1/deliveries', + '/delivery/v1/receipts', + 'IDEMPOTENT_RECEIPT_V1', + 'domovoy-notification-write-v1', + FLOOR(EXTRACT(EPOCH FROM clock_timestamp()) * 1000)::bigint, + 'AVAILABLE' +) +ON CONFLICT (binding_id, revision) DO UPDATE SET + https_origin = EXCLUDED.https_origin, + verified_at_unix_millis = EXCLUDED.verified_at_unix_millis, + health = EXCLUDED.health; + +INSERT INTO capability.notification_binding_heads ( + capability_id, owner_principal_id, binding_id, revision +) +VALUES ( + 'prospective-notification-delivery', + 'domovoy-creator-principal', + 'domovoy-notification-binding', + 2 +) +ON CONFLICT (capability_id, owner_principal_id) DO UPDATE SET + binding_id = EXCLUDED.binding_id, + revision = EXCLUDED.revision +WHERE capability.notification_binding_heads.revision < EXCLUDED.revision; + +RESET ROLE; +COMMIT; diff --git a/docs/PROJECT_STATUS.md b/docs/PROJECT_STATUS.md index 2d6745b..6237e66 100644 --- a/docs/PROJECT_STATUS.md +++ b/docs/PROJECT_STATUS.md @@ -5,7 +5,7 @@ Audit date: 2026-08-17. Active implementation: DEV-000065 revision 1. ## DEV-000065 checkpoint - `DEV_000065_STATUS = IN_PROGRESS` -- `LAST_COMPLETED = ADR-0014 accepted; Continuity-owned AttentionItem contract and persistence; due reminder scanner; Commit internal prospective notification endpoint; human-facing Chat / Today / History surface; relative reminder parsing for "через N минут"; deterministic due-delivery regression tests` +- `LAST_COMPLETED = ADR-0014 accepted; Continuity-owned AttentionItem contract and persistence; due reminder scanner; Commit internal prospective notification endpoint; human-facing Chat / Today / History surface; relative reminder parsing for "через N минут"; deterministic due-delivery regression tests; notification binding origin corrected to assistant.cin.su revision 2` - `ADR_STATUS = ADR-0014 accepted` - `OWNERSHIP_STATUS = no new owner; AttentionItem belongs to Continuity; notification delivery remains Commit/Capability/receiver path` - `REMINDER_LIFECYCLE_STATUS = ARMED -> TRIGGERED -> REVALIDATING -> DELIVERED implemented for due scanner` diff --git a/modules/continuity/ports/repository.go b/modules/continuity/ports/repository.go index 07a6738..ef2dea6 100644 --- a/modules/continuity/ports/repository.go +++ b/modules/continuity/ports/repository.go @@ -15,5 +15,6 @@ type Repository interface { PendingAttention(context.Context, int) ([]continuity.AttentionItem, error) MarkAttentionDelivering(context.Context, continuity.AttentionItemID, continuity.AttentionRevision) (continuity.AttentionItem, error) MarkAttentionDelivered(context.Context, continuity.AttentionItemID, continuity.AttentionRevision, continuity.CommittedActionIntentID, continuity.CommittedAuthorizedActionID, continuity.CommittedActionAttemptID, string, int64) (continuity.AttentionItem, error) + MarkAttentionFailed(context.Context, continuity.AttentionItemID, continuity.AttentionRevision, continuity.CommittedActionIntentID, continuity.CommittedAuthorizedActionID, continuity.CommittedActionAttemptID, string) (continuity.AttentionItem, error) AttentionByOwner(context.Context, identity.PrincipalID, int) ([]continuity.AttentionItem, error) } -- 2.54.0