94 lines
3.2 KiB
Go
94 lines
3.2 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"flag"
|
|
"fmt"
|
|
"log"
|
|
"net/http"
|
|
"os"
|
|
"time"
|
|
|
|
"git.cin.su/m/Assistant/adapters/postgres"
|
|
secretcontracts "git.cin.su/m/Assistant/contracts/secrets"
|
|
platformsecrets "git.cin.su/m/Assistant/platform/secrets"
|
|
)
|
|
|
|
const epistemicsDBPasswordPath = "/run/secrets/epistemics_db_password"
|
|
|
|
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()
|
|
password, err := resolveDatabasePassword("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()
|
|
status := newWorkerStatus(*workerID, *once)
|
|
listenAddr := os.Getenv("REFLECTION_LISTEN_ADDR")
|
|
if listenAddr == "" {
|
|
listenAddr = ":8092"
|
|
}
|
|
go func() {
|
|
if err := http.ListenAndServe(listenAddr, status.handler()); err != nil {
|
|
log.Printf("reflection status server stopped: %v", err)
|
|
}
|
|
}()
|
|
for {
|
|
now := time.Now().UnixMilli()
|
|
status.cycleStarted(time.Now())
|
|
if _, err := repository.RecoverReflectionClaims(ctx, now); err != nil {
|
|
status.cycleFailed(time.Now(), err)
|
|
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 {
|
|
status.cycleFailed(time.Now(), err)
|
|
panic(err)
|
|
}
|
|
fmt.Printf("resolved reflection candidate %s\n", candidate.ID)
|
|
} else if *once {
|
|
fmt.Println("no reflection candidate")
|
|
}
|
|
status.cycleFinished(time.Now(), map[bool]string{true: "candidate-resolved", false: "idle"}[claimErr == nil])
|
|
if *once {
|
|
return
|
|
}
|
|
if err := repository.WaitForReflectionCandidate(ctx, *pollInterval); err != nil {
|
|
status.cycleFailed(time.Now(), err)
|
|
panic(err)
|
|
}
|
|
}
|
|
}
|
|
|
|
func resolveDatabasePassword(refName string) (string, error) {
|
|
if _, ok := os.LookupEnv("EPISTEMICS_POSTGRES_PASSWORD"); ok {
|
|
return "", fmt.Errorf("EPISTEMICS_POSTGRES_PASSWORD direct environment secret is forbidden")
|
|
}
|
|
ref := os.Getenv(refName)
|
|
if ref == "" {
|
|
return "", fmt.Errorf("%s is required", 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: secretcontracts.RuntimeConsumerEpistemicsWorker, Path: epistemicsDBPasswordPath})
|
|
}
|