Compare commits

..
Author SHA1 Message Date
m 3a563a29ee feat(deploy): add independent supervisor control plane
CI / test (push) Failing after 28s
2026-08-29 12:33:44 +03:00
m 734db5d779 merge: synchronize integration with current main
CI / test (push) Failing after 28s
2026-08-29 12:25:26 +03:00
m 36fc8ec8c6 merge: resume Gen2 controlled canary
CI / test (push) Failing after 27s
2026-08-29 12:16:21 +03:00
m eeeba6b7c0 merge: Gen2 database provisioning remediation
CI / test (push) Failing after 31s
2026-08-29 12:12:43 +03:00
8 changed files with 404 additions and 13 deletions
+198 -5
View File
@@ -1,10 +1,203 @@
package main
import "net/http"
import (
"encoding/json"
"errors"
"fmt"
"net/http"
"net/http/httputil"
"net/url"
"os"
"strings"
"time"
"git.cin.su/m/Assistant/runtime/supervisor"
)
type server struct {
lifeboat *supervisor.PersistentLifeboat
stableUpstream string
candidate string
client *http.Client
}
func main() {
_ = http.ListenAndServe(":8088", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.WriteHeader(http.StatusOK)
_, _ = w.Write([]byte("lifeboat-ready\n"))
}))
statePath := envOr("SUPERVISOR_STATE_PATH", "/var/lib/supervisor/release-state.json")
stable := strings.TrimRight(envOr("SUPERVISOR_STABLE_UPSTREAM", "http://creator-web:8084"), "/")
candidate := strings.TrimRight(envOr("SUPERVISOR_CANDIDATE_UPSTREAM", "http://creator-web-candidate:8084"), "/")
store, err := supervisor.NewDurableStore(statePath)
if err != nil {
fatal(err)
}
lifeboat := supervisor.NewPersistentLifeboat(store)
if _, err := lifeboat.Snapshot(); errors.Is(err, os.ErrNotExist) {
baseline := supervisor.Release{ID: envOr("SUPERVISOR_BASELINE_RELEASE", "bootstrap"), Version: envOr("SUPERVISOR_BASELINE_RELEASE", "bootstrap")}
if ok, startErr := lifeboat.Start(baseline); startErr != nil || !ok {
fatal(fmt.Errorf("initialize baseline: %v", startErr))
}
if ok, observeErr := lifeboat.Record(supervisor.Observation{ReleaseID: baseline.ID, Ready: true, Reason: "supervisor baseline"}); observeErr != nil || !ok {
fatal(fmt.Errorf("initialize baseline observation: %v", observeErr))
}
if ok, trafficErr := lifeboat.SetTraffic(supervisor.TrafficStable, stable, candidate); trafficErr != nil || !ok {
fatal(fmt.Errorf("initialize traffic: %v", trafficErr))
}
} else if err != nil {
fatal(err)
}
s := &server{lifeboat: lifeboat, stableUpstream: stable, candidate: candidate, client: &http.Client{Timeout: 5 * time.Second}}
public := http.NewServeMux()
public.HandleFunc("GET /healthz", s.health)
public.HandleFunc("GET /readyz", s.ready)
public.HandleFunc("/", s.proxy)
control := http.NewServeMux()
control.HandleFunc("GET /v1/state", s.state)
control.HandleFunc("POST /v1/releases/start", s.start)
control.HandleFunc("POST /v1/releases/observe", s.observe)
control.HandleFunc("POST /v1/releases/promote", s.promote)
control.HandleFunc("POST /v1/releases/rollback", s.rollback)
control.HandleFunc("POST /v1/traffic/canary", s.canary)
control.HandleFunc("POST /v1/traffic/promote", s.promoteTraffic)
control.HandleFunc("POST /v1/traffic/rollback", s.rollbackTraffic)
go func() {
if err := http.ListenAndServe(envOr("SUPERVISOR_CONTROL_ADDR", ":8095"), control); err != nil {
fatal(err)
}
}()
if err := http.ListenAndServe(envOr("LISTEN_ADDR", ":8080"), public); err != nil {
fatal(err)
}
}
func (s *server) health(w http.ResponseWriter, _ *http.Request) {
writeJSON(w, map[string]string{"runtime": "supervisor", "status": "ok"})
}
func (s *server) ready(w http.ResponseWriter, r *http.Request) {
state, err := s.lifeboat.Snapshot()
if err != nil || state.StableUpstream == "" {
http.Error(w, "supervisor state unavailable", http.StatusServiceUnavailable)
return
}
request, err := http.NewRequestWithContext(r.Context(), http.MethodGet, state.StableUpstream+"/healthz", nil)
if err != nil {
http.Error(w, "stable upstream unavailable", http.StatusServiceUnavailable)
return
}
response, err := s.client.Do(request)
if err != nil || response.StatusCode/100 != 2 {
if response != nil {
response.Body.Close()
}
http.Error(w, "stable upstream unavailable", http.StatusServiceUnavailable)
return
}
response.Body.Close()
writeJSON(w, map[string]string{"runtime": "supervisor", "status": "ready"})
}
func (s *server) proxy(w http.ResponseWriter, r *http.Request) {
state, err := s.lifeboat.Snapshot()
if err != nil {
http.Error(w, "supervisor state unavailable", http.StatusServiceUnavailable)
return
}
target := state.StableUpstream
selected := "stable"
if state.TrafficMode == supervisor.TrafficCandidate || (state.TrafficMode == supervisor.TrafficCanaryHeader && strings.EqualFold(r.Header.Get("X-Assistant-Canary"), "true")) {
target = state.CandidateUpstream
selected = "candidate"
}
parsed, err := url.Parse(target)
if err != nil || parsed.Scheme == "" || parsed.Host == "" {
http.Error(w, "invalid upstream", http.StatusServiceUnavailable)
return
}
proxy := httputil.NewSingleHostReverseProxy(parsed)
proxy.ModifyResponse = func(response *http.Response) error {
response.Header.Set("X-Assistant-Traffic-Target", selected)
return nil
}
proxy.ErrorHandler = func(writer http.ResponseWriter, _ *http.Request, _ error) {
http.Error(writer, "upstream unavailable", http.StatusBadGateway)
}
proxy.ServeHTTP(w, r)
}
func (s *server) state(w http.ResponseWriter, _ *http.Request) {
state, err := s.lifeboat.Snapshot()
if err != nil {
http.Error(w, "state unavailable", 503)
return
}
writeJSON(w, state)
}
func (s *server) start(w http.ResponseWriter, r *http.Request) {
var release supervisor.Release
if !decode(w, r, &release) {
return
}
ok, err := s.lifeboat.Start(release)
respond(w, ok, err)
}
func (s *server) observe(w http.ResponseWriter, r *http.Request) {
var observation supervisor.Observation
if !decode(w, r, &observation) {
return
}
ok, err := s.lifeboat.Record(observation)
respond(w, ok, err)
}
func (s *server) promote(w http.ResponseWriter, _ *http.Request) {
ok, err := s.lifeboat.Promote()
respond(w, ok, err)
}
func (s *server) rollback(w http.ResponseWriter, _ *http.Request) {
ok, err := s.lifeboat.Rollback()
respond(w, ok, err)
}
func (s *server) canary(w http.ResponseWriter, _ *http.Request) {
ok, err := s.lifeboat.SetTraffic(supervisor.TrafficCanaryHeader, s.stableUpstream, s.candidate)
respond(w, ok, err)
}
func (s *server) promoteTraffic(w http.ResponseWriter, _ *http.Request) {
ok, err := s.lifeboat.Promote()
respond(w, ok, err)
}
func (s *server) rollbackTraffic(w http.ResponseWriter, _ *http.Request) {
ok, err := s.lifeboat.RollbackTraffic()
respond(w, ok, err)
}
func decode(w http.ResponseWriter, r *http.Request, value any) bool {
decoder := json.NewDecoder(http.MaxBytesReader(w, r.Body, 16<<10))
decoder.DisallowUnknownFields()
if err := decoder.Decode(value); err != nil {
http.Error(w, "invalid control request", 400)
return false
}
return true
}
func respond(w http.ResponseWriter, ok bool, err error) {
if err != nil {
http.Error(w, "control state unavailable", 503)
return
}
if !ok {
http.Error(w, "control transition rejected", 409)
return
}
writeJSON(w, map[string]bool{"ok": true})
}
func writeJSON(w http.ResponseWriter, value any) {
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode(value)
}
func envOr(name, fallback string) string {
if value := strings.TrimSpace(os.Getenv(name)); value != "" {
return value
}
return fallback
}
func fatal(err error) { fmt.Fprintln(os.Stderr, err); os.Exit(1) }
+31 -2
View File
@@ -277,7 +277,7 @@ services:
- --port
- "8086"
- --model
- /models/Qwen3-8B-Q4_K_M.gguf
- /models/Qwen3.8-27B-UD-Q2_K_XL.gguf
- --alias
- domovoy-lifeboat
- --jinja
@@ -299,6 +299,29 @@ services:
timeout: 3s
retries: 10
supervisor:
build: { context: .., dockerfile: deploy/images/runtime.Dockerfile, args: { TARGET: supervisor, SOURCE_REVISION: "${ASSISTANT_SOURCE_REVISION:-unknown}" } }
restart: unless-stopped
environment:
LISTEN_ADDR: ":8080"
SUPERVISOR_CONTROL_ADDR: ":8095"
SUPERVISOR_STATE_PATH: /var/lib/supervisor/release-state.json
SUPERVISOR_STABLE_UPSTREAM: http://creator-web:8084
SUPERVISOR_CANDIDATE_UPSTREAM: http://creator-web-candidate:8084
SUPERVISOR_BASELINE_RELEASE: ${ASSISTANT_SOURCE_REVISION:-unknown}
ports:
- "${CREATOR_WEB_PORT:-18080}:8080"
- "127.0.0.1:${SUPERVISOR_CONTROL_PORT:-18085}:8095"
depends_on:
creator-web: { condition: service_healthy }
volumes:
- assistant_supervisor_state:/var/lib/supervisor
healthcheck:
test: ["CMD-SHELL", "wget -qO- http://localhost:8080/healthz >/dev/null && wget -qO- http://localhost:8080/readyz >/dev/null"]
interval: 10s
timeout: 5s
retries: 20
creator-web:
build: { context: .., dockerfile: deploy/images/runtime.Dockerfile, args: { TARGET: creator-web, SOURCE_REVISION: "${ASSISTANT_SOURCE_REVISION:-unknown}" } }
restart: unless-stopped
@@ -310,7 +333,6 @@ services:
COGNITIVE_RUNTIME_URL: http://cognitive-worker:8083
AGENT_WORKER_URL: http://agent-worker:8090
ASSISTANT_BUILD_REVISION: ${ASSISTANT_SOURCE_REVISION:-unknown}
ports: ["${CREATOR_WEB_PORT:-18080}:8084"]
depends_on: [core-runtime, continuity-runtime, commit-runtime, cognitive-worker]
healthcheck:
test: ["CMD-SHELL", "wget -qO- http://localhost:8084/healthz >/dev/null"]
@@ -318,6 +340,12 @@ services:
timeout: 3s
retries: 10
creator-web-candidate:
profiles: ["assistant-canary"]
extends:
service: creator-web
restart: "no"
volumes:
assistant_pg:
assistant_nats:
@@ -333,6 +361,7 @@ volumes:
external: true
assistant_research_sources:
name: assistant_research_sources
assistant_supervisor_state:
networks:
llm-local:
+42
View File
@@ -0,0 +1,42 @@
#!/usr/bin/env bash
set -euo pipefail
BASE_DIR="${REPO_DIR:-$(CDPATH= cd -- "$(dirname -- "$0")/.." && pwd)}"
ENV_FILE="${ENV_FILE:-/home/m/assistant-deploy/production-compose.env}"
COMPOSE_FILE="$BASE_DIR/deploy/docker-compose.prod.yml"
SUPERVISOR_CONTROL_URL="${SUPERVISOR_CONTROL_URL:-http://127.0.0.1:${SUPERVISOR_CONTROL_PORT:-18085}}"
REVISION="${1:?candidate revision is required}"
set -a
. "$ENV_FILE"
set +a
compose() { docker compose --env-file "$ENV_FILE" -f "$COMPOSE_FILE" "$@"; }
control() { curl -fsS --max-time 10 -X POST "$SUPERVISOR_CONTROL_URL$1" "${@:2}" >/dev/null; }
fail_rollback() {
control /v1/traffic/rollback || true
control /v1/releases/rollback || true
echo "production canary failed; stable traffic rollback requested" >&2
exit 1
}
compose --profile assistant-canary build --build-arg "SOURCE_REVISION=$REVISION" creator-web-candidate
compose --profile assistant-canary up -d --no-build creator-web-candidate
for attempt in $(seq 1 30); do
status="$(docker inspect -f '{{if .State.Health}}{{.State.Health.Status}}{{else}}unknown{{end}}' assistant-creator-web-candidate-1 2>/dev/null || true)"
[[ "$status" == healthy ]] && break
[[ "$attempt" == 30 ]] && fail_rollback
sleep 2
done
control /v1/releases/start -H 'Content-Type: application/json' --data "{\"id\":\"candidate-$REVISION\",\"version\":\"$REVISION\"}" || fail_rollback
control /v1/releases/observe -H 'Content-Type: application/json' --data "{\"release_id\":\"candidate-$REVISION\",\"ready\":true,\"reason\":\"candidate healthcheck passed\",\"observed_at_unix_millis\":$(date +%s%3N)}" || fail_rollback
control /v1/traffic/canary || fail_rollback
headers="$(mktemp)"
trap 'rm -f "$headers"' EXIT
if ! curl -fsS --max-time 10 -D "$headers" -H 'X-Assistant-Canary: true' "https://assistant.cin.su/login" -o /dev/null; then fail_rollback; fi
grep -qi '^X-Assistant-Traffic-Target: candidate' "$headers" || fail_rollback
control /v1/releases/promote || fail_rollback
echo "production canary promoted revision=$REVISION"
+1 -1
View File
@@ -1 +1 @@
{"task_file":"dev/tasks/GEN2-027-production-canary-and-controlled-gen2-traffic-transition.json","task_id":"GEN2-027","task_revision":1}
{"task_file":"dev/tasks/GEN2-DEPLOY-001-supervisor-and-production-rollback.json","task_id":"GEN2-DEPLOY-001","task_revision":1}
@@ -0,0 +1,36 @@
{
"task_id": "GEN2-DEPLOY-001",
"task_revision": 1,
"architecture_revision": "ARCH-GEN2-000",
"base_repo_revision": "gen2-integration",
"risk_tier": "E",
"intent": "Развернуть независимый Supervisor control plane с durable release state, controlled traffic switching и доказуемым rollback перед GEN2-027.",
"requirement_ids": ["REQ-GEN2-DEPLOY-001-SUPERVISOR", "REQ-GEN2-DEPLOY-001-TRAFFIC", "REQ-GEN2-DEPLOY-001-ROLLBACK", "REQ-GEN2-DEPLOY-001-LIFEBOAT"],
"allowed_change_scope": {
"modules": ["deployment control", "runtime supervisor", "runtime health"],
"paths": [
"deploy/**",
"runtime/supervisor/**",
"runtime/health/**",
"cmd/supervisor/**",
"dev/ACTIVE_TASK.json",
"dev/tasks/GEN2-DEPLOY-001-supervisor-and-production-rollback.json",
"docs/evidence/**"
]
},
"forbidden_change_scope": {
"paths": ["GEN2-028", "database/**", "security/**", "architecture/**", "Authority/Commit bypass", "cognitive-lifeboat as Supervisor"]
},
"allowed_architecture_delta": {"new_cross_module_dependencies": 0, "new_canonical_owners": 0, "new_privileges": 1, "new_third_party_dependencies": 0, "new_database_owner_roles": 0},
"contracts": {"consume": ["GEN2-027 production canary contract", "runtime/supervisor durable state", "existing deployment update controller"], "may_change": ["deployment topology, Supervisor HTTP control contract and traffic routing only"]},
"semantic_owner": "Operations/Supervisor; not a semantic owner and not a cognitive provider.",
"canonical_state": ["durable release/current/last-known-good evidence owned by Supervisor control plane", "traffic target is Supervisor-owned operational state"],
"state_transitions": ["baseline -> candidate -> observed healthy -> header canary -> promoted OR independent rollback to last-known-good"],
"side_effects_and_privileges": ["Control API is bound only to the production-internal loopback port.", "Supervisor has no database, Authority or Commit credentials.", "Candidate traffic is explicit and rollback remains available."],
"persistence_and_migrations": ["No database changes; durable state is stored in a dedicated named volume outside candidate containers."],
"failure_retry_restart": ["Supervisor restart reloads release and traffic state from its volume.", "Candidate failure must not erase last-known-good or block stable traffic rollback.", "Lifeboat model failure remains a separately reported degraded provider condition."],
"assumptions": ["GEN2-027 resumes only after this task's production evidence; GEN2-028 is not started.", "The existing public ingress forwards to the configured creator web host port."],
"acceptance_scenarios": ["GEN2-DEPLOY-001-SCN-01: Supervisor is a separate healthy production container", "GEN2-DEPLOY-001-SCN-02: release state survives Supervisor restart", "GEN2-DEPLOY-001-SCN-03: canary header reaches candidate while default traffic stays stable", "GEN2-DEPLOY-001-SCN-04: candidate failure rolls traffic back to last-known-good", "GEN2-DEPLOY-001-SCN-05: cognitive lifeboat diagnosis and degraded role are explicit", "GEN2-DEPLOY-001-EXIT: GEN2-027 rollback prerequisites are proven"],
"required_tests": ["FAST: go test ./runtime/supervisor/... ./runtime/health/... ./cmd/supervisor/...", "FAST: go vet ./runtime/supervisor/... ./runtime/health/... ./cmd/supervisor/...", "FAST governance: preflight", "FAST scope: check-diff", "FAST independent: independent", "FAST whitespace: git diff --check", "TARGETED recovery: Supervisor restart, canary and rollback on docker.cin.su"],
"terminal_gap_states": ["BLOCKED_BY_SPEC_GAP", "BLOCKED_BY_ARCHITECTURE_GAP", "BLOCKED_BY_MISSING_CAPABILITY", "BLOCKED_BY_CONFLICT", "REQUIRES_MIGRATION_DECISION"]
}
+34 -2
View File
@@ -5,6 +5,7 @@ import (
"errors"
"os"
"path/filepath"
"sync"
)
// DurableStore is supervisor-owned external control-plane state. It is not a
@@ -63,12 +64,17 @@ func (s *DurableStore) Load() (Lifeboat, error) {
}
// PersistentLifeboat reloads before every transition and commits after it.
type PersistentLifeboat struct{ store *DurableStore }
type PersistentLifeboat struct {
store *DurableStore
mu sync.Mutex
}
func NewPersistentLifeboat(store *DurableStore) *PersistentLifeboat {
return &PersistentLifeboat{store: store}
}
func (p *PersistentLifeboat) transition(fn func(*Lifeboat) bool) (bool, error) {
p.mu.Lock()
defer p.mu.Unlock()
value, err := p.store.Load()
if err != nil && !os.IsNotExist(err) {
return false, err
@@ -91,4 +97,30 @@ func (p *PersistentLifeboat) Record(observation Observation) (bool, error) {
func (p *PersistentLifeboat) Rollback() (bool, error) {
return p.transition(func(value *Lifeboat) bool { return value.Rollback() })
}
func (p *PersistentLifeboat) Snapshot() (Lifeboat, error) { return p.store.Load() }
func (p *PersistentLifeboat) SetTraffic(mode TrafficMode, stableUpstream, candidateUpstream string) (bool, error) {
return p.transition(func(value *Lifeboat) bool { return value.SetTraffic(mode, stableUpstream, candidateUpstream) })
}
func (p *PersistentLifeboat) Promote() (bool, error) {
return p.transition(func(value *Lifeboat) bool {
if value.Current.State != ReleaseHealthy || value.Current.ID == "" {
return false
}
return value.SetTraffic(TrafficCandidate, value.StableUpstream, value.CandidateUpstream)
})
}
func (p *PersistentLifeboat) RollbackTraffic() (bool, error) {
return p.transition(func(value *Lifeboat) bool {
if !value.Rollback() {
return false
}
return value.SetTraffic(TrafficStable, value.StableUpstream, value.CandidateUpstream)
})
}
func (p *PersistentLifeboat) Snapshot() (Lifeboat, error) {
p.mu.Lock()
defer p.mu.Unlock()
return p.store.Load()
}
+27 -3
View File
@@ -9,6 +9,14 @@ const (
ReleaseRolledBack ReleaseState = "ROLLED_BACK"
)
type TrafficMode string
const (
TrafficStable TrafficMode = "STABLE"
TrafficCanaryHeader TrafficMode = "CANARY_HEADER"
TrafficCandidate TrafficMode = "CANDIDATE"
)
type Release struct {
ID string
Version string
@@ -22,9 +30,12 @@ type Observation struct {
ObservedAtUnixMillis int64
}
type Lifeboat struct {
Current Release
LastKnownGood Release
Observations []Observation
Current Release
LastKnownGood Release
Observations []Observation
TrafficMode TrafficMode
StableUpstream string
CandidateUpstream string
}
func (l *Lifeboat) Start(candidate Release) bool {
@@ -60,3 +71,16 @@ func (l *Lifeboat) Rollback() bool {
l.Current.State = ReleaseRolledBack
return true
}
func (l *Lifeboat) SetTraffic(mode TrafficMode, stableUpstream, candidateUpstream string) bool {
if mode != TrafficStable && mode != TrafficCanaryHeader && mode != TrafficCandidate {
return false
}
if stableUpstream == "" || (mode != TrafficStable && candidateUpstream == "") {
return false
}
l.TrafficMode = mode
l.StableUpstream = stableUpstream
l.CandidateUpstream = candidateUpstream
return true
}
+35
View File
@@ -47,3 +47,38 @@ func TestPersistentLifeboatRestoresLastKnownGoodAfterCandidateFailure(t *testing
t.Fatalf("state=%+v", state)
}
}
func TestPersistentSupervisorRestoresTrafficModeAfterRestart(t *testing.T) {
store, err := supervisor.NewDurableStore(filepath.Join(t.TempDir(), "control", "release-state.json"))
if err != nil {
t.Fatal(err)
}
first := supervisor.NewPersistentLifeboat(store)
if ok, err := first.Start(supervisor.Release{ID: "candidate", Version: "2"}); err != nil || !ok {
t.Fatalf("start: %v %v", ok, err)
}
if ok, err := first.Record(supervisor.Observation{ReleaseID: "candidate", Ready: true, ObservedAtUnixMillis: 1}); err != nil || !ok {
t.Fatalf("observe: %v %v", ok, err)
}
if ok, err := first.SetTraffic(supervisor.TrafficCanaryHeader, "http://stable:8084", "http://candidate:8084"); err != nil || !ok {
t.Fatalf("canary: %v %v", ok, err)
}
second := supervisor.NewPersistentLifeboat(store)
state, err := second.Snapshot()
if err != nil {
t.Fatal(err)
}
if state.TrafficMode != supervisor.TrafficCanaryHeader || state.CandidateUpstream != "http://candidate:8084" || state.LastKnownGood.ID != "candidate" {
t.Fatalf("state=%+v", state)
}
if ok, err := second.RollbackTraffic(); err != nil || !ok {
t.Fatalf("rollback traffic: %v %v", ok, err)
}
state, err = second.Snapshot()
if err != nil {
t.Fatal(err)
}
if state.TrafficMode != supervisor.TrafficStable || state.Current.State != supervisor.ReleaseRolledBack {
t.Fatalf("rollback state=%+v", state)
}
}