feat(runtime): production workers and cabinet acceptance #100

Merged
m merged 4 commits from codex/wp-400-graph-harness into main 2026-08-22 11:03:22 +00:00
11 changed files with 258 additions and 25 deletions
@@ -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,
+12
View File
@@ -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)
+48 -9
View File
@@ -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})
}
+48 -13
View File
@@ -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})
}
+3 -1
View File
@@ -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
+32
View File
@@ -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
+2 -2
View File
@@ -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
}
@@ -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"]
}
@@ -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"]
}
+4
View File
@@ -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())
@@ -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{
`<form id="login-form"`, `aria-labelledby="diagnostics-title"`, `data-page="operations"`,
`id="learning"`, `id="reflection-list"`, `id="skill-list"`, `id="settings"`,
`id="logout"`, `role="alert"`, `aria-live="polite"`,
} {
if !strings.Contains(markup, marker) {
t.Fatalf("cabinet missing %q", marker)
}
}
}
func TestCreatorCabinetRoutesSessionAndLearningThroughAuthenticatedAPI(t *testing.T) {
script, err := os.ReadFile("../../cmd/creator-web/static/assets/console.js")
if err != nil {
t.Fatal(err)
}
code := string(script)
for _, marker := range []string{
`request("/api/session/current")`, `request("/api/learning/recent?limit=8")`,
`request("/api/advisory-policy")`, `request("/api/capabilities/cognitive-proposal/health")`, `request("/api/learning/recent?limit=1")`, `sessionStorage.getItem("domovoy.creator.session")`,
`if (response.status === 401`,
} {
if !strings.Contains(code, marker) {
t.Fatalf("console missing authenticated flow %q", marker)
}
}
}
func TestCreatorCabinetResponsiveAndReducedMotionRulesExist(t *testing.T) {
style, err := os.ReadFile("../../cmd/creator-web/static/assets/console.css")
if err != nil {
t.Fatal(err)
}
css := string(style)
for _, marker := range []string{"@media (max-width: 680px)", "@media (prefers-reduced-motion: reduce)", ".learning-grid { grid-template-columns: 1fr; }"} {
if !strings.Contains(css, marker) {
t.Fatalf("console missing responsive rule %q", marker)
}
}
}