From f73c69960cd505d5904788a8a9e10da6bc05fc40 Mon Sep 17 00:00:00 2001 From: Mikhail Date: Sat, 22 Aug 2026 13:41:44 +0300 Subject: [PATCH] feat(epistemics): add durable reflection worker --- adapters/postgres/epistemics_repository.go | 51 ++++++++++++++++++- cmd/reflection-worker/main.go | 39 ++++++++++++++ contracts/epistemics/types.go | 24 +++++---- .../032_reflection_candidate_leases.sql | 19 +++++++ dev/ACTIVE_TASK.json | 2 +- dev/tasks/DEV-000135-reflection-worker.json | 23 +++++++++ .../reflection_worker_sql_test.go | 27 ++++++++++ 7 files changed, 171 insertions(+), 14 deletions(-) create mode 100644 cmd/reflection-worker/main.go create mode 100644 database/migrations/032_reflection_candidate_leases.sql create mode 100644 dev/tasks/DEV-000135-reflection-worker.json create mode 100644 tests/research_contracts/reflection_worker_sql_test.go diff --git a/adapters/postgres/epistemics_repository.go b/adapters/postgres/epistemics_repository.go index 6b4a4aa..ef8f8f9 100644 --- a/adapters/postgres/epistemics_repository.go +++ b/adapters/postgres/epistemics_repository.go @@ -64,7 +64,7 @@ func (r *EpistemicsRepository) RecentReflectionCandidates(ctx context.Context, l } rows, err := r.pool.Query(ctx, `SELECT reflection_candidate_id,trigger,state,source_experience_id,original_intent_id,source_utterance_id, - summary,evidence_refs,requires_cognition,recorded_at_unix_millis,revision + summary,evidence_refs,requires_cognition,recorded_at_unix_millis,revision,claimed_by,claim_expires_at_unix_millis FROM epistemics.reflection_candidates ORDER BY recorded_at_unix_millis DESC, reflection_candidate_id DESC LIMIT $1`, limit) if err != nil { return nil, err @@ -74,7 +74,7 @@ func (r *EpistemicsRepository) RecentReflectionCandidates(ctx context.Context, l for rows.Next() { var value epistemics.ReflectionCandidate if err := rows.Scan(&value.ID, &value.Trigger, &value.State, &value.SourceExperienceID, &value.OriginalIntentID, &value.SourceUtteranceID, - &value.Summary, &value.EvidenceRefs, &value.RequiresCognition, &value.RecordedAtUnixMillis, &value.Revision); err != nil { + &value.Summary, &value.EvidenceRefs, &value.RequiresCognition, &value.RecordedAtUnixMillis, &value.Revision, &value.ClaimedBy, &value.ClaimExpiresAtUnixMillis); err != nil { return nil, err } result = append(result, value) @@ -82,6 +82,53 @@ func (r *EpistemicsRepository) RecentReflectionCandidates(ctx context.Context, l return result, rows.Err() } +func (r *EpistemicsRepository) ClaimReflectionCandidate(ctx context.Context, workerID string, nowUnixMillis, leaseMillis int64) (epistemics.ReflectionCandidate, error) { + if workerID == "" || leaseMillis < 1 { + return epistemics.ReflectionCandidate{}, fmt.Errorf("invalid reflection claim") + } + tx, err := r.pool.Begin(ctx) + if err != nil { + return epistemics.ReflectionCandidate{}, err + } + defer tx.Rollback(ctx) + var value epistemics.ReflectionCandidate + err = tx.QueryRow(ctx, `SELECT reflection_candidate_id,trigger,state,source_experience_id,original_intent_id,source_utterance_id,summary,evidence_refs,requires_cognition,recorded_at_unix_millis,revision,claimed_by,claim_expires_at_unix_millis + FROM epistemics.reflection_candidates + WHERE state='OPEN' AND (claim_expires_at_unix_millis=0 OR claim_expires_at_unix_millis < $1) + ORDER BY recorded_at_unix_millis,reflection_candidate_id + FOR UPDATE SKIP LOCKED LIMIT 1`, nowUnixMillis).Scan(&value.ID, &value.Trigger, &value.State, &value.SourceExperienceID, &value.OriginalIntentID, &value.SourceUtteranceID, &value.Summary, &value.EvidenceRefs, &value.RequiresCognition, &value.RecordedAtUnixMillis, &value.Revision, &value.ClaimedBy, &value.ClaimExpiresAtUnixMillis) + if err != nil { + return epistemics.ReflectionCandidate{}, err + } + value.State = epistemics.ReflectionCandidateEvaluating + value.ClaimedBy = workerID + value.ClaimExpiresAtUnixMillis = nowUnixMillis + leaseMillis + _, err = tx.Exec(ctx, `UPDATE epistemics.reflection_candidates SET state=$1,claimed_by=$2,claim_expires_at_unix_millis=$3 WHERE reflection_candidate_id=$4 AND state='OPEN'`, value.State, value.ClaimedBy, value.ClaimExpiresAtUnixMillis, value.ID) + if err != nil { + return epistemics.ReflectionCandidate{}, err + } + if err := tx.Commit(ctx); err != nil { + return epistemics.ReflectionCandidate{}, err + } + return value, nil +} + +func (r *EpistemicsRepository) RecoverReflectionClaims(ctx context.Context, nowUnixMillis int64) (int64, error) { + result, err := r.pool.Exec(ctx, `UPDATE epistemics.reflection_candidates SET state='OPEN',claimed_by='',claim_expires_at_unix_millis=0 WHERE state='EVALUATING' AND claim_expires_at_unix_millis > 0 AND claim_expires_at_unix_millis < $1`, nowUnixMillis) + return result.RowsAffected(), err +} + +func (r *EpistemicsRepository) ResolveReflectionCandidate(ctx context.Context, id epistemics.ReflectionCandidateID, workerID string, nowUnixMillis int64) error { + if id == "" || workerID == "" { + return fmt.Errorf("invalid reflection resolution") + } + result, err := r.pool.Exec(ctx, `UPDATE epistemics.reflection_candidates SET state='RESOLVED',claimed_by='',claim_expires_at_unix_millis=0,revision=revision+1 WHERE reflection_candidate_id=$1 AND state='EVALUATING' AND claimed_by=$2 AND claim_expires_at_unix_millis >= $3`, id, workerID, nowUnixMillis) + if err == nil && result.RowsAffected() != 1 { + return fmt.Errorf("reflection candidate claim lost") + } + return err +} + func (r *EpistemicsRepository) SaveExperienceRecord(ctx context.Context, value epistemics.ExperienceRecord) error { _, err := r.pool.Exec(ctx, `INSERT INTO epistemics.experience_records ( experience_id, source_kind, original_intent_id, source_utterance_id, context_refs, help_method, help_method_ref, diff --git a/cmd/reflection-worker/main.go b/cmd/reflection-worker/main.go new file mode 100644 index 0000000..9a7bfcc --- /dev/null +++ b/cmd/reflection-worker/main.go @@ -0,0 +1,39 @@ +package main + +import ( + "context" + "flag" + "fmt" + "os" + "time" + + "git.cin.su/m/Assistant/adapters/postgres" +) + +func main() { + workerID := flag.String("worker-id", "reflection-worker-1", "stable worker identity") + lease := flag.Duration("lease-ttl", 5*time.Minute, "candidate claim lease") + flag.Parse() + ctx := context.Background() + config := postgres.ContinuityConfig{Host: os.Getenv("POSTGRES_HOST"), Port: os.Getenv("POSTGRES_PORT"), Database: os.Getenv("POSTGRES_DATABASE"), User: os.Getenv("EPISTEMICS_POSTGRES_USER"), Password: os.Getenv("EPISTEMICS_POSTGRES_PASSWORD")} + repository, err := postgres.OpenEpistemics(ctx, config) + if err != nil { + panic(err) + } + defer repository.Close() + now := time.Now().UnixMilli() + if _, err := repository.RecoverReflectionClaims(ctx, now); err != nil { + panic(err) + } + candidate, err := repository.ClaimReflectionCandidate(ctx, *workerID, now, lease.Milliseconds()) + if err != nil { + fmt.Println("no reflection candidate") + return + } + // This worker only closes the durable evaluation loop. Any cognitive + // proposal remains a candidate and must be admitted by a separate governed flow. + if err := repository.ResolveReflectionCandidate(ctx, candidate.ID, *workerID, time.Now().UnixMilli()); err != nil { + panic(err) + } + fmt.Printf("resolved reflection candidate %s\n", candidate.ID) +} diff --git a/contracts/epistemics/types.go b/contracts/epistemics/types.go index da7be41..2927aed 100644 --- a/contracts/epistemics/types.go +++ b/contracts/epistemics/types.go @@ -120,15 +120,17 @@ type ExperienceRecord struct { // Cognitive reflection, if used, must go through Cognitive Gateway/Router and // produce candidates only; it cannot directly rewrite canonical state. type ReflectionCandidate struct { - ID ReflectionCandidateID - Trigger ReflectionTriggerKind - State ReflectionCandidateState - SourceExperienceID ExperienceRecordID - OriginalIntentID OriginalIntentRef - SourceUtteranceID SourceUtteranceRef - Summary string - EvidenceRefs []EvidenceRef - RequiresCognition bool - RecordedAtUnixMillis int64 - Revision int64 + ID ReflectionCandidateID + Trigger ReflectionTriggerKind + State ReflectionCandidateState + SourceExperienceID ExperienceRecordID + OriginalIntentID OriginalIntentRef + SourceUtteranceID SourceUtteranceRef + Summary string + EvidenceRefs []EvidenceRef + RequiresCognition bool + RecordedAtUnixMillis int64 + Revision int64 + ClaimedBy string + ClaimExpiresAtUnixMillis int64 } diff --git a/database/migrations/032_reflection_candidate_leases.sql b/database/migrations/032_reflection_candidate_leases.sql new file mode 100644 index 0000000..be110b6 --- /dev/null +++ b/database/migrations/032_reflection_candidate_leases.sql @@ -0,0 +1,19 @@ +BEGIN; + +SET ROLE epistemics_role; + +ALTER TABLE epistemics.reflection_candidates + ADD COLUMN IF NOT EXISTS claimed_by TEXT NOT NULL DEFAULT '', + ADD COLUMN IF NOT EXISTS claim_expires_at_unix_millis BIGINT NOT NULL DEFAULT 0; + +ALTER TABLE epistemics.reflection_candidates + DROP CONSTRAINT IF EXISTS reflection_candidates_claim_expiry_check; +ALTER TABLE epistemics.reflection_candidates + ADD CONSTRAINT reflection_candidates_claim_expiry_check + CHECK (claim_expires_at_unix_millis >= 0); + +CREATE INDEX IF NOT EXISTS reflection_candidates_claimable_idx + ON epistemics.reflection_candidates (state, claim_expires_at_unix_millis, recorded_at_unix_millis, reflection_candidate_id); + +RESET ROLE; +COMMIT; diff --git a/dev/ACTIVE_TASK.json b/dev/ACTIVE_TASK.json index bd408de..673e0bf 100644 --- a/dev/ACTIVE_TASK.json +++ b/dev/ACTIVE_TASK.json @@ -1,5 +1,5 @@ { "task_file": "dev/tasks/DEV-000134-truth-bundle-compiler.json", - "task_id": "DEV-000134", + "task_id": "DEV-000135", "task_revision": 1 } diff --git a/dev/tasks/DEV-000135-reflection-worker.json b/dev/tasks/DEV-000135-reflection-worker.json new file mode 100644 index 0000000..64b04c8 --- /dev/null +++ b/dev/tasks/DEV-000135-reflection-worker.json @@ -0,0 +1,23 @@ +{ + "task_id": "DEV-000135", + "task_revision": 1, + "architecture_revision": "ADR-0018", + "base_repo_revision": "32e33e3", + "risk_tier": "E", + "intent": "Add durable lease, stale recovery, and one-shot governed closure for reflection candidates.", + "requirement_ids": ["WP-501", "Reflection candidate worker", "Stale claim recovery"], + "allowed_change_scope": {"modules": ["epistemics", "epistemics-storage"], "paths": ["contracts/epistemics/types.go", "database/migrations/032_reflection_candidate_leases.sql", "adapters/postgres/epistemics_repository.go", "cmd/reflection-worker/main.go", "tests/research_contracts/reflection_worker_sql_test.go", "dev/ACTIVE_TASK.json", "dev/tasks/DEV-000135-reflection-worker.json"]}, + "forbidden_change_scope": {"paths": ["architecture/**", "security/**", "deploy/**", "contracts/authority/**", "modules/cognitive/**", ".gitea/**"]}, + "allowed_architecture_delta": {"new_cross_module_dependencies": 0, "new_canonical_owners": 0, "new_privileges": 1, "new_third_party_dependencies": 0, "new_database_owner_roles": 1}, + "contracts": {"consume": ["Existing epistemics ReflectionCandidate contract"], "may_change": ["ReflectionCandidate lease metadata"]}, + "semantic_owner": "epistemics", + "canonical_state": ["epistemics.reflection_candidates"], + "state_transitions": ["OPEN to EVALUATING under lease; EVALUATING to OPEN on expiry; EVALUATING to RESOLVED by owning worker"], + "side_effects_and_privileges": ["Epistemics schema only; no cognitive call and no proposal promotion"], + "persistence_and_migrations": ["Append-only migration 032; production apply is later canary"], + "failure_retry_restart": ["FOR UPDATE SKIP LOCKED claim; expired claims are recoverable; one-shot worker is restart-safe"], + "assumptions": ["Cognitive evaluation and skill admission remain separate governed stages"], + "acceptance_scenarios": ["Concurrent workers claim distinct candidates", "expired evaluation returns to OPEN", "only claim owner can resolve"], + "required_tests": ["go test ./...", "go vet ./...", "preflight", "check-diff", "check-architecture", "independent", "git diff --check"], + "terminal_gap_states": ["BLOCKED_BY_MIGRATION_DECISION", "BLOCKED_BY_SPEC_GAP", "BLOCKED_BY_CONFLICT"] +} diff --git a/tests/research_contracts/reflection_worker_sql_test.go b/tests/research_contracts/reflection_worker_sql_test.go new file mode 100644 index 0000000..461f909 --- /dev/null +++ b/tests/research_contracts/reflection_worker_sql_test.go @@ -0,0 +1,27 @@ +package research_contracts_test + +import ( + "os" + "strings" + "testing" +) + +func TestReflectionCandidateLeaseSQLIsDurableAndRecoverable(t *testing.T) { + checks := map[string][]string{ + "../../database/migrations/032_reflection_candidate_leases.sql": {"claim_expires_at_unix_millis", "reflection_candidates_claimable_idx"}, + "../../adapters/postgres/epistemics_repository.go": {"claim_expires_at_unix_millis", "FOR UPDATE SKIP LOCKED", "state='OPEN'", "state='RESOLVED'"}, + "../../cmd/reflection-worker/main.go": {"RecoverReflectionClaims", "ClaimReflectionCandidate", "ResolveReflectionCandidate"}, + } + for path, needles := range checks { + data, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read %s: %v", path, err) + } + text := string(data) + for _, needle := range needles { + if !strings.Contains(text, needle) { + t.Fatalf("%s missing %q", path, needle) + } + } + } +}