diff --git a/adapters/postgres/epistemics_repository.go b/adapters/postgres/epistemics_repository.go index ef8f8f9..2992978 100644 --- a/adapters/postgres/epistemics_repository.go +++ b/adapters/postgres/epistemics_repository.go @@ -3,6 +3,7 @@ package postgres import ( "context" "fmt" + "time" "git.cin.su/m/Assistant/contracts/epistemics" "github.com/jackc/pgx/v5/pgxpool" @@ -129,6 +130,14 @@ func (r *EpistemicsRepository) ResolveReflectionCandidate(ctx context.Context, i return err } +func (r *EpistemicsRepository) WaitForReflectionCandidate(ctx context.Context, interval time.Duration) error { + if interval <= 0 { + return fmt.Errorf("invalid reflection poll interval") + } + _, err := r.pool.Exec(ctx, `SELECT pg_sleep($1)`, interval.Seconds()) + 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/adapters/postgres/work_repository.go b/adapters/postgres/work_repository.go index 38e39ce..008c226 100644 --- a/adapters/postgres/work_repository.go +++ b/adapters/postgres/work_repository.go @@ -7,6 +7,7 @@ import ( "time" "git.cin.su/m/Assistant/contracts/work" + "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" ) @@ -91,6 +92,9 @@ func (r *WorkRepository) ClaimNextDispatch(ctx context.Context, workerID string, var item work.AgentDispatchQueueItem err = tx.QueryRow(ctx, `SELECT dispatch_id,state,available_at_unix_millis,attempts,claimed_by,updated_at_unix_millis FROM work.agent_dispatch_queue WHERE state='QUEUED' AND available_at_unix_millis <= $1 ORDER BY available_at_unix_millis,dispatch_id FOR UPDATE SKIP LOCKED LIMIT 1`, nowUnixMillis).Scan(&item.DispatchID, &item.State, &item.AvailableAtUnixMillis, &item.Attempts, &item.ClaimedBy, &item.UpdatedAtUnixMillis) if err != nil { + if errors.Is(err, pgx.ErrNoRows) { + return work.AgentDispatchQueueItem{}, nil + } return work.AgentDispatchQueueItem{}, err } item.State = work.AgentDispatchClaimed @@ -107,6 +111,14 @@ func (r *WorkRepository) ClaimNextDispatch(ctx context.Context, workerID string, return item, nil } +func (r *WorkRepository) WaitForDispatch(ctx context.Context, interval time.Duration) error { + if interval <= 0 { + return fmt.Errorf("invalid worker poll interval") + } + _, err := r.pool.Exec(ctx, `SELECT pg_sleep($1)`, interval.Seconds()) + return err +} + func (r *WorkRepository) CompleteDispatch(ctx context.Context, dispatchID, workerID string, state work.AgentDispatchQueueState, nowUnixMillis int64) error { if state != work.AgentDispatchCompleted && state != work.AgentDispatchFailed { return fmt.Errorf("invalid terminal dispatch state: %s", state) diff --git a/cmd/agent-worker/main.go b/cmd/agent-worker/main.go index d793637..55a5bd4 100644 --- a/cmd/agent-worker/main.go +++ b/cmd/agent-worker/main.go @@ -9,29 +9,68 @@ import ( "time" "git.cin.su/m/Assistant/adapters/postgres" + secretcontracts "git.cin.su/m/Assistant/contracts/secrets" "git.cin.su/m/Assistant/contracts/work" "git.cin.su/m/Assistant/modules/work/application" + platformsecrets "git.cin.su/m/Assistant/platform/secrets" ) func main() { workerID := flag.String("worker-id", "agent-worker-1", "durable worker identity") staleAfter := flag.Duration("stale-after", 10*time.Minute, "age after which a claimed item is recovered") leaseTTL := flag.Duration("lease-ttl", 15*time.Minute, "dry-run lease duration") + pollInterval := flag.Duration("poll-interval", 5*time.Second, "database-owned idle wake-up interval") + once := flag.Bool("once", false, "run one cycle and exit") flag.Parse() - repository, err := postgres.OpenWork(context.Background(), postgres.ContinuityConfig{Host: os.Getenv("POSTGRES_HOST"), Port: os.Getenv("POSTGRES_PORT"), Database: os.Getenv("POSTGRES_DATABASE"), User: os.Getenv("WORK_POSTGRES_USER"), Password: os.Getenv("WORK_POSTGRES_PASSWORD")}) + password, err := resolveDatabasePassword("WORK_POSTGRES_PASSWORD", "WORK_POSTGRES_CREDENTIAL_REF", secretcontracts.RuntimeConsumerWorkWorker) + if err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(1) + } + repository, err := postgres.OpenWork(context.Background(), postgres.ContinuityConfig{Host: os.Getenv("POSTGRES_HOST"), Port: os.Getenv("POSTGRES_PORT"), Database: os.Getenv("POSTGRES_DATABASE"), User: os.Getenv("WORK_POSTGRES_USER"), Password: password}) if err != nil { fmt.Fprintln(os.Stderr, err) os.Exit(1) } defer repository.Close() - result, err := application.ExecuteWorkerCycle(context.Background(), repository, *workerID, work.AgentFencingToken(1), time.Now().UTC(), *staleAfter, *leaseTTL) - if err != nil { - fmt.Fprintln(os.Stderr, err) - os.Exit(1) - } - if err := json.NewEncoder(os.Stdout).Encode(result); err != nil { - fmt.Fprintln(os.Stderr, err) - os.Exit(1) + ctx := context.Background() + for { + result, cycleErr := application.ExecuteWorkerCycle(ctx, repository, *workerID, work.AgentFencingToken(1), time.Now().UTC(), *staleAfter, *leaseTTL) + if cycleErr != nil { + fmt.Fprintln(os.Stderr, cycleErr) + if *once { + os.Exit(1) + } + } + if cycleErr == nil && (result.Completed || result.RecoveredClaims > 0) { + _ = json.NewEncoder(os.Stdout).Encode(result) + } + if *once { + return + } + if err := repository.WaitForDispatch(ctx, *pollInterval); err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(1) + } } } + +func resolveDatabasePassword(name, refName string, consumer secretcontracts.RuntimeConsumer) (string, error) { + if _, ok := os.LookupEnv(name); ok { + return "", fmt.Errorf("%s direct environment secret is forbidden", name) + } + path, ref := os.Getenv(name+"_FILE"), os.Getenv(refName) + if path == "" || ref == "" { + return "", fmt.Errorf("%s_FILE and %s are required", name, refName) + } + root := os.Getenv("SECRET_BACKEND_ROOT") + if root == "" { + root = "/run/secrets" + } + resolver, err := platformsecrets.NewFileResolver(root) + if err != nil { + return "", err + } + return resolver.ResolveFile(secretcontracts.SecretFileRef{CredentialRef: secretcontracts.CredentialRef(ref), Purpose: secretcontracts.SecretPurposeDatabasePassword, Consumer: consumer, Path: path}) +} diff --git a/cmd/reflection-worker/main.go b/cmd/reflection-worker/main.go index 9a7bfcc..1898cd5 100644 --- a/cmd/reflection-worker/main.go +++ b/cmd/reflection-worker/main.go @@ -8,32 +8,67 @@ import ( "time" "git.cin.su/m/Assistant/adapters/postgres" + secretcontracts "git.cin.su/m/Assistant/contracts/secrets" + platformsecrets "git.cin.su/m/Assistant/platform/secrets" ) func main() { workerID := flag.String("worker-id", "reflection-worker-1", "stable worker identity") lease := flag.Duration("lease-ttl", 5*time.Minute, "candidate claim lease") + pollInterval := flag.Duration("poll-interval", 15*time.Second, "database-owned idle wake-up interval") + once := flag.Bool("once", false, "run one cycle and exit") 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")} + password, err := resolveDatabasePassword("EPISTEMICS_POSTGRES_PASSWORD", "EPISTEMICS_POSTGRES_CREDENTIAL_REF") + if err != nil { + panic(err) + } + 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: 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) + for { + now := time.Now().UnixMilli() + if _, err := repository.RecoverReflectionClaims(ctx, now); err != nil { + panic(err) + } + candidate, claimErr := repository.ClaimReflectionCandidate(ctx, *workerID, now, lease.Milliseconds()) + if claimErr == nil { + // 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) + } else if *once { + fmt.Println("no reflection candidate") + } + if *once { + return + } + if err := repository.WaitForReflectionCandidate(ctx, *pollInterval); err != nil { + panic(err) + } } - candidate, err := repository.ClaimReflectionCandidate(ctx, *workerID, now, lease.Milliseconds()) +} + +func resolveDatabasePassword(name, refName string) (string, error) { + if _, ok := os.LookupEnv(name); ok { + return "", fmt.Errorf("%s direct environment secret is forbidden", name) + } + path, ref := os.Getenv(name+"_FILE"), os.Getenv(refName) + if path == "" || ref == "" { + return "", fmt.Errorf("%s_FILE and %s are required", name, refName) + } + root := os.Getenv("SECRET_BACKEND_ROOT") + if root == "" { + root = "/run/secrets" + } + resolver, err := platformsecrets.NewFileResolver(root) 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) + return "", err } - fmt.Printf("resolved reflection candidate %s\n", candidate.ID) + return resolver.ResolveFile(secretcontracts.SecretFileRef{CredentialRef: secretcontracts.CredentialRef(ref), Purpose: secretcontracts.SecretPurposeDatabasePassword, Consumer: secretcontracts.RuntimeConsumerEpistemicsWorker, Path: path}) } diff --git a/contracts/secrets/types.go b/contracts/secrets/types.go index e599ff0..f1022b6 100644 --- a/contracts/secrets/types.go +++ b/contracts/secrets/types.go @@ -15,7 +15,9 @@ const ( type RuntimeConsumer string const ( - RuntimeConsumerCore RuntimeConsumer = "core-runtime" + RuntimeConsumerCore RuntimeConsumer = "core-runtime" + RuntimeConsumerWorkWorker RuntimeConsumer = "agent-worker" + RuntimeConsumerEpistemicsWorker RuntimeConsumer = "reflection-worker" ) // SecretFileRef is bootstrap delivery metadata for the V1 protected local diff --git a/deploy/docker-compose.prod.yml b/deploy/docker-compose.prod.yml index 4478d7b..f5dcf94 100644 --- a/deploy/docker-compose.prod.yml +++ b/deploy/docker-compose.prod.yml @@ -97,6 +97,38 @@ services: WORK_POSTGRES_PASSWORD: ${WORK_DB_PASSWORD} depends_on: { postgres: { condition: service_healthy } } + agent-worker: + build: { context: .., dockerfile: deploy/images/runtime.Dockerfile, args: { TARGET: agent-worker, SOURCE_REVISION: "${ASSISTANT_SOURCE_REVISION:-unknown}" } } + restart: unless-stopped + command: ["-worker-id", "agent-worker-production"] + environment: + POSTGRES_HOST: postgres + POSTGRES_PORT: "5432" + POSTGRES_DATABASE: ${POSTGRES_DB:-assistant} + WORK_POSTGRES_USER: work_role + WORK_POSTGRES_CREDENTIAL_REF: credential://domovoy/work/db-password/v1 + WORK_POSTGRES_PASSWORD_FILE: /run/secrets/work_db_password + SECRET_BACKEND_ROOT: /run/secrets + depends_on: { postgres: { condition: service_healthy } } + volumes: + - assistant_core_secrets:/run/secrets:ro + + reflection-worker: + build: { context: .., dockerfile: deploy/images/runtime.Dockerfile, args: { TARGET: reflection-worker, SOURCE_REVISION: "${ASSISTANT_SOURCE_REVISION:-unknown}" } } + restart: unless-stopped + command: ["-worker-id", "reflection-worker-production"] + environment: + POSTGRES_HOST: postgres + POSTGRES_PORT: "5432" + POSTGRES_DATABASE: ${POSTGRES_DB:-assistant} + EPISTEMICS_POSTGRES_USER: epistemics_role + EPISTEMICS_POSTGRES_CREDENTIAL_REF: credential://domovoy/epistemics/db-password/v1 + EPISTEMICS_POSTGRES_PASSWORD_FILE: /run/secrets/epistemics_db_password + SECRET_BACKEND_ROOT: /run/secrets + depends_on: { postgres: { condition: service_healthy } } + volumes: + - assistant_core_secrets:/run/secrets:ro + continuity-runtime: build: { context: .., dockerfile: deploy/images/runtime.Dockerfile, args: { TARGET: continuity-runtime, SOURCE_REVISION: "${ASSISTANT_SOURCE_REVISION:-unknown}" } } restart: unless-stopped diff --git a/dev/ACTIVE_TASK.json b/dev/ACTIVE_TASK.json index 56e69b3..efa3daf 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-000137", + "task_file": "dev/tasks/DEV-000139-production-worker-runtime.json", + "task_id": "DEV-000139", "task_revision": 1 } diff --git a/dev/tasks/DEV-000138-ui-operations-acceptance-contract.json b/dev/tasks/DEV-000138-ui-operations-acceptance-contract.json new file mode 100644 index 0000000..978983a --- /dev/null +++ b/dev/tasks/DEV-000138-ui-operations-acceptance-contract.json @@ -0,0 +1,23 @@ +{ + "task_id": "DEV-000138", + "task_revision": 1, + "architecture_revision": "ADR-0018", + "base_repo_revision": "1ccd43c", + "risk_tier": "E", + "intent": "Add executable contract coverage for authenticated cabinet, operations, learning, accessibility, responsive, and reduced-motion surfaces.", + "requirement_ids": ["WP-801", "WP-802", "WP-805", "Authenticated cabinet acceptance"], + "allowed_change_scope": {"modules": ["creator-web"], "paths": ["tests/creator_web/cabinet_acceptance_contract_test.go", "dev/ACTIVE_TASK.json", "dev/tasks/DEV-000138-ui-operations-acceptance-contract.json"]}, + "forbidden_change_scope": {"paths": ["architecture/**", "security/**", "deploy/**", "database/**", "cmd/**", "contracts/**", ".gitea/**"]}, + "allowed_architecture_delta": {"new_cross_module_dependencies": 0, "new_canonical_owners": 0, "new_privileges": 0, "new_third_party_dependencies": 0, "new_database_owner_roles": 0}, + "contracts": {"consume": ["Existing creator-web authenticated API and static cabinet"], "may_change": ["Acceptance tests only"]}, + "semantic_owner": "creator-web", + "canonical_state": ["None; tests inspect existing UI contract"], + "state_transitions": ["login session gates cabinet; operations and learning request authenticated API data"], + "side_effects_and_privileges": ["No external mutation; live authenticated browser acceptance remains a separate production verification"], + "persistence_and_migrations": ["None"], + "failure_retry_restart": ["Missing auth, learning, or responsive contract fails CI"], + "assumptions": ["Credentials are never embedded in acceptance tests"], + "acceptance_scenarios": ["cabinet exposes operations and reflection surfaces", "session and learning routes are authenticated", "mobile/reduced motion rules are present"], + "required_tests": ["go test ./...", "go vet ./...", "preflight", "check-diff", "check-architecture", "independent", "git diff --check"], + "terminal_gap_states": ["BLOCKED_BY_AUTHENTICATED_BROWSER_SESSION", "BLOCKED_BY_CONFLICT"] +} diff --git a/dev/tasks/DEV-000139-production-worker-runtime.json b/dev/tasks/DEV-000139-production-worker-runtime.json new file mode 100644 index 0000000..7872b58 --- /dev/null +++ b/dev/tasks/DEV-000139-production-worker-runtime.json @@ -0,0 +1,23 @@ +{ + "task_id": "DEV-000139", + "task_revision": 1, + "architecture_revision": "ADR-0018", + "base_repo_revision": "6f5f7e1", + "risk_tier": "E", + "intent": "Run agent and reflection workers as restartable production processes with DB-owned idle wakeups and protected credential refs.", + "requirement_ids": ["WP-302", "WP-502", "Automatic worker runtime", "Production recovery"], + "allowed_change_scope": {"modules": ["work-agent-execution", "epistemics", "deployment"], "paths": ["contracts/secrets/types.go", "modules/work/application/worker.go", "adapters/postgres/work_repository.go", "cmd/agent-worker/main.go", "adapters/postgres/epistemics_repository.go", "cmd/reflection-worker/main.go", "deploy/docker-compose.prod.yml", "dev/ACTIVE_TASK.json", "dev/tasks/DEV-000139-production-worker-runtime.json"]}, + "forbidden_change_scope": {"paths": ["architecture/**", "security/**", "database/**", "contracts/authority/**", ".gitea/**"]}, + "allowed_architecture_delta": {"new_cross_module_dependencies": 0, "new_canonical_owners": 0, "new_privileges": 2, "new_third_party_dependencies": 0, "new_database_owner_roles": 0}, + "contracts": {"consume": ["Existing durable dispatch/reflection tables and file secret resolver"], "may_change": ["Worker process lifecycle and production service wiring"]}, + "semantic_owner": "work-agent-execution and epistemics", + "canonical_state": ["work.agent_dispatch_queue", "epistemics.reflection_candidates"], + "state_transitions": ["workers recover stale claims, claim durable work, process one cycle, and repeat after DB-owned wakeup"], + "side_effects_and_privileges": ["Work and epistemics database roles only; credentials remain CredentialRef/file-backed"], + "persistence_and_migrations": ["Uses migrations 030 and 032 already in main"], + "failure_retry_restart": ["restartable containers; stale claims recover after restart; -once remains diagnostic mode"], + "assumptions": ["Production secret volume contains work and epistemics refs at declared paths"], + "acceptance_scenarios": ["both workers stay running idle", "worker survives empty queue", "worker uses protected secret files", "stale dispatch/reflection claims recover"], + "required_tests": ["go test ./...", "go vet ./...", "preflight", "check-diff", "check-architecture", "independent", "git diff --check", "docker compose config"], + "terminal_gap_states": ["BLOCKED_BY_SECRET_REF_MISMATCH", "BLOCKED_BY_MIGRATION_DECISION", "BLOCKED_BY_CONFLICT"] +} diff --git a/modules/work/application/worker.go b/modules/work/application/worker.go index 62638b9..b687a94 100644 --- a/modules/work/application/worker.go +++ b/modules/work/application/worker.go @@ -12,6 +12,7 @@ type WorkerCycleResult struct { RecoveredClaims int64 DispatchID string Completed bool + Idle bool } func ExecuteWorkerCycle(ctx context.Context, store ports.WorkerStore, workerID string, token work.AgentFencingToken, now time.Time, staleAfter, leaseTTL time.Duration) (WorkerCycleResult, error) { @@ -23,6 +24,9 @@ func ExecuteWorkerCycle(ctx context.Context, store ports.WorkerStore, workerID s if err != nil { return WorkerCycleResult{RecoveredClaims: recovered}, err } + if item.DispatchID == "" { + return WorkerCycleResult{RecoveredClaims: recovered, Idle: true}, nil + } intent, err := store.LoadDispatchIntent(ctx, item.DispatchID) if err != nil { _ = store.CompleteDispatch(ctx, item.DispatchID, workerID, work.AgentDispatchFailed, now.UnixMilli()) diff --git a/tests/creator_web/cabinet_acceptance_contract_test.go b/tests/creator_web/cabinet_acceptance_contract_test.go new file mode 100644 index 0000000..b0d5965 --- /dev/null +++ b/tests/creator_web/cabinet_acceptance_contract_test.go @@ -0,0 +1,54 @@ +package creator_web_test + +import ( + "os" + "strings" + "testing" +) + +func TestCreatorCabinetHasAuthenticatedOperationalAndLearningSurfaces(t *testing.T) { + page, err := os.ReadFile("../../cmd/creator-web/static/index.html") + if err != nil { + t.Fatal(err) + } + markup := string(page) + for _, marker := range []string{ + `