feat(deploy): add independent supervisor control plane

adapters/multimodal/ingress.go

@@ -0,0 +1,18 @@
+package multimodal
+
+import (
+	"crypto/sha256"
+	"encoding/hex"
+	"git.cin.su/m/Assistant/contracts/artifact"
+	"git.cin.su/m/Assistant/contracts/capability"
+	"git.cin.su/m/Assistant/contracts/interaction"
+)
+
+func NormalizeIngress(id string, modality interaction.IngressModality, sourceRef, deviceRef, locationRef, principalID, contentType string, data []byte, response capability.MultimodalResponse, at int64) (interaction.IngressEnvelope, artifact.MediaArtifact) {
+	hash := sha256.Sum256(data)
+	artifactValue := artifact.MediaArtifact{ID: artifact.ArtifactID("artifact:media:" + hex.EncodeToString(hash[:8])), ContentType: contentType, SizeBytes: int64(len(data)), ContentDigest: hex.EncodeToString(hash[:]), SourceRef: sourceRef, RecordedAtUnixMillis: at}
+	if len(data) == 0 {
+		return interaction.IngressEnvelope{ID: id, Modality: modality, SourceRef: sourceRef, DeviceRef: deviceRef, LocationRef: locationRef, PrincipalID: principalID, Degraded: true, RecordedAtUnixMillis: at}, artifact.MediaArtifact{}
+	}
+	return interaction.IngressEnvelope{ID: id, Modality: modality, SourceRef: sourceRef, DeviceRef: deviceRef, LocationRef: locationRef, PrincipalID: principalID, ArtifactRef: string(artifactValue.ID), ObservationRef: response.EvidenceRef, ProviderRef: response.Provider, ModelRef: response.Model, ModelVersion: response.ModelVersion, ConfidenceMillis: response.ConfidenceMillis, RecordedAtUnixMillis: at}, artifactValue
+}

adapters/multimodal/ingress_test.go

@@ -0,0 +1,22 @@
+package multimodal_test
+
+import (
+	"git.cin.su/m/Assistant/adapters/multimodal"
+	"git.cin.su/m/Assistant/contracts/capability"
+	"git.cin.su/m/Assistant/contracts/interaction"
+	"testing"
+)
+
+func TestIngressNormalizesMediaArtifactAndObservationProvenance(t *testing.T) {
+	e, a := multimodal.NormalizeIngress("ingress-1", interaction.IngressImage, "camera:street", "device:cam", "world:street", "principal", "image/png", []byte("image"), capability.MultimodalResponse{Provider: "local", Model: "vision", ModelVersion: "1", EvidenceRef: "observation-1", ConfidenceMillis: 800}, 1)
+	if !e.Valid() || !a.Valid() || e.ArtifactRef == "" || e.ObservationRef != "observation-1" || e.DeviceRef == "" || e.LocationRef == "" {
+		t.Fatalf("envelope=%+v artifact=%+v", e, a)
+	}
+}
+
+func TestUnavailableIngressDegradesWithoutFabricatingObservation(t *testing.T) {
+	e, a := multimodal.NormalizeIngress("ingress-1", interaction.IngressAudio, "mic", "device:mic", "", "principal", "audio/wav", nil, capability.MultimodalResponse{}, 1)
+	if !e.Degraded || e.ObservationRef != "" || a.Valid() {
+		t.Fatalf("envelope=%+v artifact=%+v", e, a)
+	}
+}

cmd/core-runtime/main.go

@@ -533,7 +533,7 @@ func main() {
 			log.Printf("interaction principal history loaded: %d utterances", len(recentUserHistory))
 		}
 		conversationHistory = append(conversationHistory, recentUserHistory...)
-		gen2Context := interactiondomain.BuildGen2Context(conversationHistory)
+		gen2Context := interactiondomain.BuildScopedGen2Context(conversationHistory, string(session.PrincipalID))
 		if !creativeUserRequest && locationReply == "" {
 			if continuation := interactiondomain.ContinueLocalSearch(effectiveTaskMessage, conversationHistory); continuation != "" {
 				continuationMessage = continuation

cmd/supervisor/main.go

@@ -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) {
+	statePath := envOr("SUPERVISOR_STATE_PATH", "/var/lib/supervisor/release-state.json")
-		w.WriteHeader(http.StatusOK)
+	stable := strings.TrimRight(envOr("SUPERVISOR_STABLE_UPSTREAM", "http://creator-web:8084"), "/")
-		_, _ = w.Write([]byte("lifeboat-ready\n"))
+	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) }

contracts/artifact/types.go

@@ -5,6 +5,18 @@ import "git.cin.su/m/Assistant/platform/ids"
 import "errors"
 
 type ArtifactID ids.ID
+type MediaArtifact struct {
+	ID                   ArtifactID
+	ContentType          string
+	SizeBytes            int64
+	ContentDigest        string
+	SourceRef            string
+	RecordedAtUnixMillis int64
+}
+
+func (a MediaArtifact) Valid() bool {
+	return a.ID != "" && a.ContentType != "" && a.SizeBytes > 0 && a.ContentDigest != "" && a.SourceRef != "" && a.RecordedAtUnixMillis > 0
+}
 
 type AgentArtifactAttestation struct {
 	ArtifactID          ArtifactID

contracts/capability/multimodal.go

@@ -15,11 +15,15 @@ const (
 )
 
 type MultimodalResponse struct {
-	Operation      MultimodalOperation `json:"operation"`
+	Operation        MultimodalOperation `json:"operation"`
-	Text           string              `json:"text"`
+	Text             string              `json:"text"`
-	Provider       string              `json:"provider,omitempty"`
+	Provider         string              `json:"provider,omitempty"`
-	ObservedAt     int64               `json:"observed_at_unix_millis"`
+	ObservedAt       int64               `json:"observed_at_unix_millis"`
-	ConfidenceNote string              `json:"confidence_note,omitempty"`
+	ConfidenceNote   string              `json:"confidence_note,omitempty"`
-	ContentType    string              `json:"content_type,omitempty"`
+	ContentType      string              `json:"content_type,omitempty"`
-	DataBase64     string              `json:"data_base64,omitempty"`
+	DataBase64       string              `json:"data_base64,omitempty"`
+	Model            string              `json:"model,omitempty"`
+	ModelVersion     string              `json:"model_version,omitempty"`
+	ConfidenceMillis int64               `json:"confidence_millis,omitempty"`
+	EvidenceRef      string              `json:"evidence_ref,omitempty"`
 }

contracts/interaction/types.go

@@ -33,6 +33,36 @@ type SpeakerAttribution struct {
 	Authenticated      bool
 	ObservationRef     string
 }
+type IngressModality string
+
+const (
+	IngressText   IngressModality = "TEXT"
+	IngressAudio  IngressModality = "AUDIO"
+	IngressImage  IngressModality = "IMAGE"
+	IngressVideo  IngressModality = "VIDEO"
+	IngressDevice IngressModality = "DEVICE"
+)
+
+type IngressEnvelope struct {
+	ID                   string
+	Modality             IngressModality
+	SourceRef            string
+	DeviceRef            string
+	LocationRef          string
+	PrincipalID          string
+	ArtifactRef          string
+	ObservationRef       string
+	ProviderRef          string
+	ModelRef             string
+	ModelVersion         string
+	ConfidenceMillis     int64
+	Degraded             bool
+	RecordedAtUnixMillis int64
+}
+
+func (e IngressEnvelope) Valid() bool {
+	return e.ID != "" && e.Modality != "" && e.SourceRef != "" && e.ConfidenceMillis >= 0 && e.ConfidenceMillis <= 1000 && e.RecordedAtUnixMillis > 0 && (!e.Degraded || e.ObservationRef == "")
+}
 
 const (
 	ChannelWebChat                               Channel                         = "WEB_CHAT"

database/bootstrap/002_gen2_roles_and_schemas.sql

@@ -0,0 +1,58 @@
+-- Gen2 owner-role provisioning for the dedicated bootstrap/migration principal.
+-- This is intentionally separate from feature migrations and may be run repeatedly.
+BEGIN;
+
+DO $$
+BEGIN
+  IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'world_role') THEN
+    CREATE ROLE world_role LOGIN;
+  END IF;
+  IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'configuration_role') THEN
+    CREATE ROLE configuration_role LOGIN;
+  END IF;
+  IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'self_role') THEN
+    CREATE ROLE self_role LOGIN;
+  END IF;
+END
+$$;
+
+DO $$
+DECLARE
+  role_name TEXT;
+  can_login BOOLEAN;
+BEGIN
+  FOREACH role_name IN ARRAY ARRAY['world_role', 'configuration_role', 'self_role'] LOOP
+    SELECT rolcanlogin INTO can_login FROM pg_roles WHERE rolname = role_name;
+    IF can_login IS DISTINCT FROM TRUE THEN
+      RAISE EXCEPTION 'Gen2 owner role % exists but is not a V1-compatible LOGIN role', role_name;
+    END IF;
+  END LOOP;
+END
+$$;
+
+CREATE SCHEMA IF NOT EXISTS world AUTHORIZATION world_role;
+CREATE SCHEMA IF NOT EXISTS configuration AUTHORIZATION configuration_role;
+CREATE SCHEMA IF NOT EXISTS self AUTHORIZATION self_role;
+
+REVOKE ALL ON SCHEMA world, configuration, self FROM PUBLIC;
+
+DO $$
+DECLARE
+  schema_name TEXT;
+  expected_owner TEXT;
+  actual_owner TEXT;
+BEGIN
+  FOREACH schema_name IN ARRAY ARRAY['world', 'configuration', 'self'] LOOP
+    expected_owner := schema_name || '_role';
+    SELECT pg_get_userbyid(nspowner)
+      INTO actual_owner
+      FROM pg_namespace
+     WHERE nspname = schema_name;
+    IF actual_owner IS DISTINCT FROM expected_owner THEN
+      RAISE EXCEPTION 'Gen2 schema % is owned by %, expected %', schema_name, actual_owner, expected_owner;
+    END IF;
+  END LOOP;
+END
+$$;
+
+COMMIT;

deploy/apply-migrations.sh

@@ -25,6 +25,16 @@ psql_exec() {
   docker exec -i "$POSTGRES_CONTAINER" psql -v ON_ERROR_STOP=1 -U "$POSTGRES_SUPERUSER" -d "$DATABASE" "$@"
 }
 
+# Owner roles and schemas are provisioned by the dedicated bootstrap boundary,
+# before feature migrations run. This keeps role creation out of migrations and
+# makes existing PostgreSQL volumes converge idempotently as well.
+GEN2_BOOTSTRAP="$REPO_DIR/database/bootstrap/002_gen2_roles_and_schemas.sql"
+if [[ ! -f "$GEN2_BOOTSTRAP" ]]; then
+  echo "migration runner: Gen2 bootstrap is missing" >&2
+  exit 1
+fi
+psql_exec < "$GEN2_BOOTSTRAP"
+
 psql_exec -c "CREATE TABLE IF NOT EXISTS public.assistant_migration_history (migration_id TEXT PRIMARY KEY, applied_at_unix_millis BIGINT NOT NULL CHECK (applied_at_unix_millis > 0));"
 
 if [[ -n "${MIGRATION_BASELINE:-}" ]]; then

deploy/docker-compose.prod.yml

@@ -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:

deploy/images/postgres.Dockerfile

@@ -4,5 +4,6 @@ LABEL org.opencontainers.image.revision=$SOURCE_REVISION \
       org.opencontainers.image.source="https://git.cin.su/m/Assistant.git"
 
 COPY database/bootstrap/001_roles_and_schemas.sql /docker-entrypoint-initdb.d/001_roles_and_schemas.sql
+COPY database/bootstrap/002_gen2_roles_and_schemas.sql /docker-entrypoint-initdb.d/002_gen2_roles_and_schemas.sql
 COPY deploy/images/002_owner_passwords.sh /docker-entrypoint-initdb.d/002_owner_passwords.sh
 RUN chmod 0555 /docker-entrypoint-initdb.d/002_owner_passwords.sh

deploy/run-production-canary.sh

@@ -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"

dev/ACTIVE_TASK.json

@@ -1 +1 @@
-{"task_file":"dev/tasks/GEN2-024-multi-person-realms-identity-boundaries-and-contextual-source-trust.json","task_id":"GEN2-024","task_revision":1}
+{"task_file":"dev/tasks/GEN2-DEPLOY-001-supervisor-and-production-rollback.json","task_id":"GEN2-DEPLOY-001","task_revision":1}

dev/tasks/GEN2-DB-001-provision-gen2-owner-roles.json

@@ -0,0 +1,42 @@
+{
+  "task_id": "GEN2-DB-001",
+  "task_revision": 1,
+  "architecture_revision": "ARCH-GEN2-000",
+  "base_repo_revision": "gen2-integration",
+  "risk_tier": "E",
+  "intent": "Исправить provisioning Gen2 owner roles/schemas до применения migrations 046-050 без ослабления V1 DB isolation.",
+  "requirement_ids": ["REQ-GEN2-DB-001-ROLES", "REQ-GEN2-DB-001-OWNERSHIP", "REQ-GEN2-DB-001-IDEMPOTENCY", "REQ-GEN2-DB-001-RECOVERY"],
+  "allowed_change_scope": {
+    "modules": ["database bootstrap", "deployment migration runner"],
+    "paths": [
+      "database/bootstrap/002_gen2_roles_and_schemas.sql",
+      "deploy/apply-migrations.sh",
+      "deploy/images/postgres.Dockerfile",
+      "dev/tasks/GEN2-DB-001-provision-gen2-owner-roles.json",
+      "dev/ACTIVE_TASK.json",
+      "dev/tasks/GEN2-DB-001-provision-gen2-owner-roles.json"
+    ],
+  },
+  "forbidden_change_scope": {
+    "paths": [
+      "GEN2-028",
+      "production-canary-before-provisioning",
+      "feature-migration-role-creation",
+      "cross-owner-grants",
+      "new canonical owner",
+      "production canary before provisioning"
+    ]
+  },
+  "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": ["V1 database ownership/ACL", "GEN2 canonical owner registry", "GEN2 execution contract"], "may_change": ["Bootstrap provisioning only; feature migration semantics are unchanged."]},
+  "semantic_owner": "Dedicated PostgreSQL bootstrap/migration principal; semantic ownership remains World, Configuration, SELF and existing V1 owners.",
+  "canonical_state": ["world schema owned by world_role", "configuration schema owned by configuration_role", "self schema owned by self_role", "no cross-owner canonical writes"],
+  "state_transitions": ["missing Gen2 roles/schemas -> idempotently provisioned -> migrations 046-050 apply -> restart/recovery preserves schema and migration state"],
+  "side_effects_and_privileges": ["Create only the three declared LOGIN owner roles when absent.", "Revoke PUBLIC schema privileges and fail closed on an unexpected schema owner.", "Do not invent or print credentials; do not grant cross-owner access."],
+  "persistence_and_migrations": ["Provisioning runs in the dedicated bootstrap boundary before feature migrations.", "Migrations 046-050 remain unchanged and are applied by the existing migration history runner."],
+  "failure_retry_restart": ["Provisioning is safe to run repeatedly.", "Unexpected existing role/schema ownership fails closed and requires an architecture decision.", "PostgreSQL restart/recovery must retain owner roles, schemas, ACL and migration history."],
+  "assumptions": ["This is a user-authorized remediation task for the previously stopped GEN2-027 migration decision.", "No Gen2 runtime consumer currently requires a new database password; credentials are not invented by this task.", "GEN2-027 resumes only after provisioning and recovery evidence; GEN2-028 is not started."],
+  "acceptance_scenarios": ["GEN2-DB-001-SCN-01: all roles required by migrations 046-050 are identified", "GEN2-DB-001-SCN-02: provisioning succeeds twice without duplicate objects", "GEN2-DB-001-SCN-03: migrations 046-050 apply in the controlled deployment path", "GEN2-DB-001-SCN-04: schema owners and PUBLIC ACL are confirmed", "GEN2-DB-001-SCN-05: PostgreSQL restart/recovery preserves the result", "GEN2-DB-001-EXIT: migration blocker is removed without weakening V1 isolation"],
+  "required_tests": ["FAST governance: go run ./cmd/devctl preflight -task dev/tasks/GEN2-DB-001-provision-gen2-owner-roles.json", "FAST scope: go run ./cmd/devctl check-diff -active dev/ACTIVE_TASK.json", "FAST whitespace: git diff --check gen2-integration --", "TARGETED: provisioning repeated twice, migrations 046-050, ACL/ownership verification and PostgreSQL restart/recovery"],
+  "terminal_gap_states": ["BLOCKED_BY_SPEC_GAP", "BLOCKED_BY_ARCHITECTURE_GAP", "BLOCKED_BY_MISSING_CAPABILITY", "BLOCKED_BY_CONFLICT", "REQUIRES_MIGRATION_DECISION"]
+}

dev/tasks/GEN2-DEPLOY-001-supervisor-and-production-rollback.json

@@ -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"]
+}

modules/interaction/domain/gen2_context.go

@@ -11,6 +11,23 @@ type ContextProjection struct {
 // BuildGen2Context is a read-side projection over canonical utterance history.
 // PersonalMemory remains a compatibility fallback and is never written here.
 func BuildGen2Context(history []interaction.HumanUtterance) ContextProjection {
+	return BuildScopedGen2Context(history, "")
+}
+
+// BuildScopedGen2Context is the Gen2 retrieval boundary for household
+// context. The caller supplies already-authorized history; this defensive
+// filter prevents accidental cross-principal projection when a mixed source
+// is passed during degraded operation.
+func BuildScopedGen2Context(history []interaction.HumanUtterance, principalID string) ContextProjection {
+	if principalID != "" {
+		scoped := make([]interaction.HumanUtterance, 0, len(history))
+		for _, item := range history {
+			if item.PrincipalID == "" || string(item.PrincipalID) == principalID {
+				scoped = append(scoped, item)
+			}
+		}
+		history = scoped
+	}
 	text := CompilePersonalMemory(history).ContextText()
 	if text != "" {
 		return ContextProjection{Text: text, Source: "retrieval:interaction-utterances"}

modules/interaction/domain/gen2_context_test.go

@@ -3,6 +3,7 @@ package domain_test
 import (
 	"git.cin.su/m/Assistant/contracts/interaction"
 	"git.cin.su/m/Assistant/modules/interaction/domain"
+	"strings"
 	"testing"
 )
 
@@ -16,3 +17,13 @@ func TestGen2ContextUsesRetrievalProjectionAndKeepsV1Fallback(t *testing.T) {
 		t.Fatalf("fallback missing: %#v", empty)
 	}
 }
+
+func TestGen2ContextDoesNotProjectAnotherPrincipal(t *testing.T) {
+	got := domain.BuildScopedGen2Context([]interaction.HumanUtterance{
+		{ID: "mine", PrincipalID: "principal-1", Content: "Я сейчас в Москве", ReceivedAtUnixMillis: 2},
+		{ID: "other", PrincipalID: "principal-2", Content: "Я сейчас в Казани", ReceivedAtUnixMillis: 3},
+	}, "principal-1")
+	if got.Text == "" || strings.Contains(got.Text, "Казани") || !strings.Contains(got.Text, "Москве") {
+		t.Fatalf("scoped context=%#v", got)
+	}
+}

runtime/supervisor/durable_store.go

@@ -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()
+}

runtime/supervisor/supervisor.go

@@ -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
+	Current           Release
-	LastKnownGood Release
+	LastKnownGood     Release
-	Observations  []Observation
+	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
+}

runtime/supervisor/supervisor_test.go

@@ -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)
+	}
+}