From ebcae4cc9c1360a8ddb6a945ed88fe5074c1fe95 Mon Sep 17 00:00:00 2001 From: Mikhail Date: Sat, 22 Aug 2026 11:57:43 +0300 Subject: [PATCH 1/2] Add cognitive admission policy --- dev/ACTIVE_TASK.json | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/dev/ACTIVE_TASK.json b/dev/ACTIVE_TASK.json index 104f028..5694a52 100644 --- a/dev/ACTIVE_TASK.json +++ b/dev/ACTIVE_TASK.json @@ -1,5 +1,5 @@ { - "task_file": "dev/tasks/DEV-000140-research-evaluation-runtime.json", - "task_id": "DEV-000140", + "task_file": "dev/tasks/DEV-000115-cognitive-admission-control.json", + "task_id": "DEV-000115", "task_revision": 1 } -- 2.54.0 From 4c4a85aec6dcd13f33ecf43e3bf693272a1af914 Mon Sep 17 00:00:00 2001 From: Mikhail Date: Sat, 22 Aug 2026 14:33:53 +0300 Subject: [PATCH 2/2] feat(cognitive): enforce durable admission leases --- adapters/postgres/capability_repository.go | 83 +++++++++++++++++++ cmd/cognitive-worker/admission.go | 68 +++++++++++++++ cmd/cognitive-worker/admission_test.go | 49 +++++++++++ cmd/cognitive-worker/gateway.go | 30 ++++++- cmd/cognitive-worker/main.go | 2 +- cmd/core-runtime/main.go | 36 ++++++++ contracts/capability/admission.go | 5 ++ .../033_cognitive_admission_leases.sql | 30 +++++++ deploy/docker-compose.prod.yml | 5 ++ dev/ACTIVE_TASK.json | 2 +- ...EV-000115-cognitive-admission-control.json | 32 +++---- 11 files changed, 320 insertions(+), 22 deletions(-) create mode 100644 cmd/cognitive-worker/admission.go create mode 100644 cmd/cognitive-worker/admission_test.go create mode 100644 database/migrations/033_cognitive_admission_leases.sql diff --git a/adapters/postgres/capability_repository.go b/adapters/postgres/capability_repository.go index 1d62375..ce30e63 100644 --- a/adapters/postgres/capability_repository.go +++ b/adapters/postgres/capability_repository.go @@ -28,6 +28,89 @@ func OpenCapability(ctx context.Context, config ContinuityConfig) (*CapabilityRe func (r *CapabilityRepository) Close() { r.pool.Close() } +func (r *CapabilityRepository) AdmitCognitive(ctx context.Context, leaseID string, policy capability.AdmissionPolicy, request capability.AdmissionRequest, nowUnixMillis int64) (capability.AdmissionDecision, capability.AdmissionLease, error) { + tx, err := r.pool.Begin(ctx) + if err != nil { + return capability.AdmissionDecision{}, capability.AdmissionLease{}, err + } + defer func() { _ = tx.Rollback(ctx) }() + if leaseID == "" || nowUnixMillis < 1 { + return capability.AdmissionDecision{State: capability.AdmissionRejected, Reason: capability.AdmissionReasonInputBudget, PolicyID: policy.ID}, capability.AdmissionLease{}, nil + } + var existingState string + var existingPolicy string + err = tx.QueryRow(ctx, `SELECT state, policy_id FROM capability.cognitive_admission_leases WHERE lease_id=$1`, leaseID).Scan(&existingState, &existingPolicy) + if err == nil { + if existingState == "ACTIVE" { + return capability.AdmissionDecision{State: capability.AdmissionAdmitted, Reason: capability.AdmissionReasonNone, PolicyID: existingPolicy}, capability.AdmissionLease{ID: leaseID, PolicyID: existingPolicy}, tx.Commit(ctx) + } + return capability.AdmissionDecision{State: capability.AdmissionRejected, Reason: capability.AdmissionReasonConcurrency, PolicyID: existingPolicy}, capability.AdmissionLease{}, tx.Commit(ctx) + } + if !errors.Is(err, pgx.ErrNoRows) { + return capability.AdmissionDecision{}, capability.AdmissionLease{}, err + } + windowStart := nowUnixMillis - 60*60*1000 + _, err = tx.Exec(ctx, `INSERT INTO capability.cognitive_admission_counters(policy_id,window_start_unix_millis,accepted_count,concurrent_count,updated_at_unix_millis) VALUES($1,$2,0,0,$3) ON CONFLICT(policy_id) DO NOTHING`, policy.ID, windowStart, nowUnixMillis) + if err != nil { + return capability.AdmissionDecision{}, capability.AdmissionLease{}, err + } + var storedWindow int64 + var accepted, concurrent int + if err = tx.QueryRow(ctx, `SELECT window_start_unix_millis,accepted_count,concurrent_count FROM capability.cognitive_admission_counters WHERE policy_id=$1 FOR UPDATE`, policy.ID).Scan(&storedWindow, &accepted, &concurrent); err != nil { + return capability.AdmissionDecision{}, capability.AdmissionLease{}, err + } + if nowUnixMillis-storedWindow >= 60*60*1000 { + storedWindow, accepted = windowStart, 0 + } + var expired int64 + if err = tx.QueryRow(ctx, `WITH expired AS (UPDATE capability.cognitive_admission_leases SET state='EXPIRED' WHERE policy_id=$1 AND state='ACTIVE' AND expires_at_unix_millis <= $2 RETURNING lease_id) SELECT count(*) FROM expired`, policy.ID, nowUnixMillis).Scan(&expired); err != nil { + return capability.AdmissionDecision{}, capability.AdmissionLease{}, err + } + concurrent -= int(expired) + if concurrent < 0 { + concurrent = 0 + } + decision := capability.Admit(request, policy, capability.AdmissionCounters{Concurrent: concurrent, AcceptedLastHour: accepted}, nowUnixMillis) + if decision.State != capability.AdmissionAdmitted { + if _, err = tx.Exec(ctx, `UPDATE capability.cognitive_admission_counters SET window_start_unix_millis=$2,accepted_count=$3,concurrent_count=$4,updated_at_unix_millis=$5 WHERE policy_id=$1`, policy.ID, storedWindow, accepted, concurrent, nowUnixMillis); err != nil { + return capability.AdmissionDecision{}, capability.AdmissionLease{}, err + } + return decision, capability.AdmissionLease{}, tx.Commit(ctx) + } + expires := nowUnixMillis + 60*1000 + if request.DeadlineMillis > nowUnixMillis && request.DeadlineMillis < expires { + expires = request.DeadlineMillis + } + if _, err = tx.Exec(ctx, `UPDATE capability.cognitive_admission_counters SET window_start_unix_millis=$2,accepted_count=$3,concurrent_count=$4,updated_at_unix_millis=$5 WHERE policy_id=$1`, policy.ID, storedWindow, accepted+1, concurrent+1, nowUnixMillis); err != nil { + return capability.AdmissionDecision{}, capability.AdmissionLease{}, err + } + if _, err = tx.Exec(ctx, `INSERT INTO capability.cognitive_admission_leases(lease_id,policy_id,state,expires_at_unix_millis,created_at_unix_millis) VALUES($1,$2,'ACTIVE',$3,$4)`, leaseID, policy.ID, expires, nowUnixMillis); err != nil { + return capability.AdmissionDecision{}, capability.AdmissionLease{}, err + } + return decision, capability.AdmissionLease{ID: leaseID, PolicyID: policy.ID}, tx.Commit(ctx) +} + +func (r *CapabilityRepository) ReleaseCognitiveAdmission(ctx context.Context, lease capability.AdmissionLease, nowUnixMillis int64) error { + tx, err := r.pool.Begin(ctx) + if err != nil { + return err + } + defer func() { _ = tx.Rollback(ctx) }() + var policyID string + err = tx.QueryRow(ctx, `UPDATE capability.cognitive_admission_leases SET state='RELEASED' WHERE lease_id=$1 AND state='ACTIVE' RETURNING policy_id`, lease.ID).Scan(&policyID) + if errors.Is(err, pgx.ErrNoRows) { + return tx.Commit(ctx) + } + if err != nil { + return err + } + _, err = tx.Exec(ctx, `UPDATE capability.cognitive_admission_counters SET concurrent_count=GREATEST(0,concurrent_count-1),updated_at_unix_millis=$2 WHERE policy_id=$1`, policyID, nowUnixMillis) + if err != nil { + return err + } + return tx.Commit(ctx) +} + func (r *CapabilityRepository) SaveHealth(ctx context.Context, health capability.Health) error { _, err := r.pool.Exec(ctx, `INSERT INTO capability.health (capability_id, state, reason, revision, semantic_version, observed_at_unix_millis) VALUES ($1,$2,$3,$4,$5,$6) ON CONFLICT (capability_id) DO UPDATE SET state=EXCLUDED.state, reason=EXCLUDED.reason, revision=EXCLUDED.revision, semantic_version=EXCLUDED.semantic_version, observed_at_unix_millis=EXCLUDED.observed_at_unix_millis WHERE capability.health.revision < EXCLUDED.revision`, health.CapabilityID, health.State, health.Reason, health.Revision, health.SemanticVersion, health.ObservedAtUnixMillis) return err diff --git a/cmd/cognitive-worker/admission.go b/cmd/cognitive-worker/admission.go new file mode 100644 index 0000000..ccf4f3a --- /dev/null +++ b/cmd/cognitive-worker/admission.go @@ -0,0 +1,68 @@ +package main + +import ( + "bytes" + "context" + "encoding/json" + "fmt" + "net/http" + "strings" + + "git.cin.su/m/Assistant/contracts/capability" +) + +type httpAdmissionClient struct { + baseURL string + client *http.Client +} + +func newHTTPAdmissionClient(baseURL string, client *http.Client) *httpAdmissionClient { + return &httpAdmissionClient{baseURL: strings.TrimRight(baseURL, "/"), client: client} +} + +func (c *httpAdmissionClient) Admit(ctx context.Context, leaseID string, request capability.AdmissionRequest) (capability.AdmissionDecision, capability.AdmissionLease, error) { + payload, err := json.Marshal(struct { + LeaseID string `json:"lease_id"` + Request capability.AdmissionRequest `json:"request"` + }{leaseID, request}) + if err != nil { + return capability.AdmissionDecision{}, capability.AdmissionLease{}, err + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.baseURL+"/v1/internal/cognitive/admission", bytes.NewReader(payload)) + if err != nil { + return capability.AdmissionDecision{}, capability.AdmissionLease{}, err + } + req.Header.Set("Content-Type", "application/json") + resp, err := c.client.Do(req) + if err != nil { + return capability.AdmissionDecision{}, capability.AdmissionLease{}, err + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return capability.AdmissionDecision{}, capability.AdmissionLease{}, fmt.Errorf("admission status %d", resp.StatusCode) + } + var result struct { + Decision capability.AdmissionDecision `json:"decision"` + Lease capability.AdmissionLease `json:"lease"` + } + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return capability.AdmissionDecision{}, capability.AdmissionLease{}, err + } + return result.Decision, result.Lease, nil +} + +func (c *httpAdmissionClient) Release(ctx context.Context, lease capability.AdmissionLease) error { + req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.baseURL+"/v1/internal/cognitive/admission/"+lease.ID+"/release", nil) + if err != nil { + return err + } + resp, err := c.client.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusNoContent { + return fmt.Errorf("admission release status %d", resp.StatusCode) + } + return nil +} diff --git a/cmd/cognitive-worker/admission_test.go b/cmd/cognitive-worker/admission_test.go new file mode 100644 index 0000000..f58a104 --- /dev/null +++ b/cmd/cognitive-worker/admission_test.go @@ -0,0 +1,49 @@ +package main + +import ( + "context" + "testing" + + "git.cin.su/m/Assistant/contracts/capability" + "git.cin.su/m/Assistant/contracts/cognitive" +) + +type admissionTestProvider struct{ calls int } + +func (p *admissionTestProvider) Advise(context.Context, cognitive.AdvisoryRequest) (cognitive.AdvisoryResponse, error) { + p.calls++ + return cognitive.AdvisoryResponse{FinishReason: cognitive.FinishCompleted}, nil +} + +type admissionTestClient struct { + decision capability.AdmissionDecision + releases int +} + +func (c *admissionTestClient) Admit(context.Context, string, capability.AdmissionRequest) (capability.AdmissionDecision, capability.AdmissionLease, error) { + return c.decision, capability.AdmissionLease{ID: "lease-1", PolicyID: "policy-1"}, nil +} +func (c *admissionTestClient) Release(context.Context, capability.AdmissionLease) error { + c.releases++ + return nil +} + +func TestInvokeAdmittedDoesNotCallProviderWhenRejected(t *testing.T) { + provider := &admissionTestProvider{} + client := &admissionTestClient{decision: capability.AdmissionDecision{State: capability.AdmissionRejected, Reason: capability.AdmissionReasonConcurrency, PolicyID: "policy-1"}} + gateway := cognitiveGateway{admission: client} + response, err := gateway.invokeAdmitted(context.Background(), "request-1", provider, cognitive.AdvisoryRequest{Message: "hello"}, 100) + if err != nil || response.FinishReason != cognitive.FinishRateLimited || provider.calls != 0 || client.releases != 0 { + t.Fatalf("rejected admission leaked provider call: response=%#v err=%v calls=%d releases=%d", response, err, provider.calls, client.releases) + } +} + +func TestInvokeAdmittedReleasesDurableLeaseAfterProvider(t *testing.T) { + provider := &admissionTestProvider{} + client := &admissionTestClient{decision: capability.AdmissionDecision{State: capability.AdmissionAdmitted, Reason: capability.AdmissionReasonNone, PolicyID: "policy-1"}} + gateway := cognitiveGateway{admission: client} + response, err := gateway.invokeAdmitted(context.Background(), "request-1", provider, cognitive.AdvisoryRequest{Message: "hello"}, 100) + if err != nil || response.FinishReason != cognitive.FinishCompleted || provider.calls != 1 || client.releases != 1 { + t.Fatalf("admitted lease was not released: response=%#v err=%v calls=%d releases=%d", response, err, provider.calls, client.releases) + } +} diff --git a/cmd/cognitive-worker/gateway.go b/cmd/cognitive-worker/gateway.go index a9b105c..aa9d05f 100644 --- a/cmd/cognitive-worker/gateway.go +++ b/cmd/cognitive-worker/gateway.go @@ -26,6 +26,11 @@ type bindingResolver interface { RegisterDiscovery(context.Context, capability.CognitiveDiscoverySnapshot) (capability.CognitiveDiscoveryRegistration, error) } +type admissionClient interface { + Admit(context.Context, string, capability.AdmissionRequest) (capability.AdmissionDecision, capability.AdmissionLease, error) + Release(context.Context, capability.AdmissionLease) error +} + type providerRegistration struct { ProviderInstanceID capability.ProviderInstanceID Kind capability.ProviderKind @@ -46,6 +51,7 @@ type cognitiveGateway struct { now func() time.Time allowCanary bool canaryMaxOutputBytes int + admission admissionClient } func (g cognitiveGateway) policyHandler(w http.ResponseWriter, r *http.Request) { @@ -95,9 +101,9 @@ func (g cognitiveGateway) advisoryHandler(w http.ResponseWriter, r *http.Request } invocation := cognitive.AdvisoryRequest{Version: request.Version, RequestID: request.RequestID, ActorID: request.ActorID, PrincipalID: request.PrincipalID, ScopeID: request.ScopeID, Message: request.Message, Context: request.Context, Locale: request.Locale, MaxOutputBytes: maxOutputBytes, ProviderBindingID: selected.Binding.ID, ProviderBindingRevision: selected.Binding.Revision, ProviderInstanceID: selected.Provider.ID, ModelDescriptorID: selected.Model.ID, ProviderModelID: selected.Model.ProviderModelID} started := g.now() - response, invokeErr := provider.Advise(r.Context(), invocation) + response, invokeErr := g.invokeAdmitted(r.Context(), string(request.RequestID), provider, invocation, maxOutputBytes) if invokeErr != nil { - response = cognitive.AdvisoryResponse{Version: request.Version, RequestID: request.RequestID, ProviderBindingID: selected.Binding.ID, ProviderBindingRevision: selected.Binding.Revision, FinishReason: cognitive.FinishInvalidProviderResponse} + response = cognitive.AdvisoryResponse{Version: request.Version, RequestID: request.RequestID, ProviderBindingID: selected.Binding.ID, ProviderBindingRevision: selected.Binding.Revision, FinishReason: cognitive.FinishProviderUnavailable} } g.observe(r.Context(), selected, response, started) if fallbackEligible(response.FinishReason) { @@ -109,9 +115,9 @@ func (g cognitiveGateway) advisoryHandler(w http.ResponseWriter, r *http.Request fallbackRequest.ProviderBindingID, fallbackRequest.ProviderBindingRevision = fallback.Binding.ID, fallback.Binding.Revision fallbackRequest.ProviderInstanceID, fallbackRequest.ModelDescriptorID, fallbackRequest.ProviderModelID = fallback.Provider.ID, fallback.Model.ID, fallback.Model.ProviderModelID fallbackStarted := g.now() - fallbackResponse, fallbackErr := fallbackProvider.Advise(r.Context(), fallbackRequest) + fallbackResponse, fallbackErr := g.invokeAdmitted(r.Context(), string(request.RequestID)+":fallback", fallbackProvider, fallbackRequest, maxOutputBytes) if fallbackErr != nil { - fallbackResponse = cognitive.AdvisoryResponse{Version: request.Version, RequestID: request.RequestID, ProviderBindingID: fallback.Binding.ID, ProviderBindingRevision: fallback.Binding.Revision, FinishReason: cognitive.FinishInvalidProviderResponse} + fallbackResponse = cognitive.AdvisoryResponse{Version: request.Version, RequestID: request.RequestID, ProviderBindingID: fallback.Binding.ID, ProviderBindingRevision: fallback.Binding.Revision, FinishReason: cognitive.FinishProviderUnavailable} } g.observe(r.Context(), fallback, fallbackResponse, fallbackStarted) response = fallbackResponse @@ -124,6 +130,22 @@ func (g cognitiveGateway) advisoryHandler(w http.ResponseWriter, r *http.Request writeGatewayResponse(w, cognitive.GatewayAdvisoryResponse{Version: response.Version, RequestID: response.RequestID, AdvisoryText: response.AdvisoryText, ReferencedExcerptIDs: response.ReferencedExcerptIDs, FinishReason: response.FinishReason, Usage: response.Usage}) } +func (g cognitiveGateway) invokeAdmitted(ctx context.Context, leaseID string, provider advisoryProvider, invocation cognitive.AdvisoryRequest, maxOutputBytes int) (cognitive.AdvisoryResponse, error) { + var lease capability.AdmissionLease + if g.admission != nil { + decision, admittedLease, err := g.admission.Admit(ctx, leaseID, capability.AdmissionRequest{InputBytes: len(invocation.Message), MaxOutputBytes: maxOutputBytes, DeadlineMillis: 0}) + if err != nil { + return cognitive.AdvisoryResponse{}, err + } + if decision.State != capability.AdmissionAdmitted { + return cognitive.AdvisoryResponse{Version: invocation.Version, RequestID: invocation.RequestID, ProviderBindingID: invocation.ProviderBindingID, ProviderBindingRevision: invocation.ProviderBindingRevision, FinishReason: cognitive.FinishRateLimited}, nil + } + lease = admittedLease + defer func() { _ = g.admission.Release(ctx, lease) }() + } + return provider.Advise(ctx, invocation) +} + func selectionObjectiveForMessage(message string) capability.SelectionObjective { lower := strings.ToLower(message) if strings.Contains(lower, "дешев") || strings.Contains(lower, "бесплат") || strings.Contains(lower, "эконом") || strings.Contains(lower, "cost") || strings.Contains(lower, "cheap") { diff --git a/cmd/cognitive-worker/main.go b/cmd/cognitive-worker/main.go index 3b9cdd9..3544630 100644 --- a/cmd/cognitive-worker/main.go +++ b/cmd/cognitive-worker/main.go @@ -70,7 +70,7 @@ func main() { kind := capability.ProviderKind(envOr("OPENAI_PROVIDER_KIND", "openai-responses-v1")) providers[kind] = append(providers[kind], providerRegistration{ProviderInstanceID: "openai-bootstrap", Kind: kind, EndpointRef: capability.EndpointBindingRef(envOr("OPENAI_ENDPOINT_REF", "openai-responses-default")), CredentialRef: capability.CredentialRef(envOr("OPENAI_CREDENTIAL_REF", "openai-advisory-v1")), AuthMode: capability.AuthKey, BillingMode: capability.BillingPerInput, DataPolicy: capability.DataPolicyExternalMessageOnly, Provider: openAIProvider}) } - gateway := cognitiveGateway{resolver: resolver, providers: providers, builtinLocal: builtinLocal, externalRecovery: externalRecovery, now: time.Now, allowCanary: os.Getenv("COGNITIVE_ALLOW_CANARY") == "true", canaryMaxOutputBytes: 4096} + gateway := cognitiveGateway{resolver: resolver, providers: providers, builtinLocal: builtinLocal, externalRecovery: externalRecovery, now: time.Now, allowCanary: os.Getenv("COGNITIVE_ALLOW_CANARY") == "true", canaryMaxOutputBytes: 4096, admission: newHTTPAdmissionClient(envOr("CORE_RUNTIME_URL", "http://core-runtime:8080"), &http.Client{Timeout: 5 * time.Second})} mux.Handle("/healthz", health.Handler{Runtime: "cognitive-worker", SemanticGeneration: 1, ArchitectureGeneration: 1}) readinessClient := &http.Client{Timeout: 2 * time.Second} lifeboatURL := strings.TrimSuffix(strings.TrimRight(envOr("LOCAL_LIFEBOAT_PROVIDER_URL", "http://cognitive-lifeboat:8086/v1"), "/"), "/v1") + "/health" diff --git a/cmd/core-runtime/main.go b/cmd/core-runtime/main.go index f7319d4..56ffb84 100644 --- a/cmd/core-runtime/main.go +++ b/cmd/core-runtime/main.go @@ -141,6 +141,34 @@ func main() { health.HTTPProbe(readinessClient, strings.TrimRight(envOr("CONTINUITY_RUNTIME_URL", "http://continuity-runtime:8081"), "/")+"/healthz"), health.HTTPProbe(readinessClient, strings.TrimRight(envOr("COGNITIVE_RUNTIME_URL", "http://cognitive-worker:8083"), "/")+"/healthz"), }}) + admissionPolicy := capability.AdmissionPolicy{ID: envOr("COGNITIVE_ADMISSION_POLICY_ID", "creator-advisory-local-v1"), MaxConcurrent: envInt("COGNITIVE_ADMISSION_MAX_CONCURRENT", 1), MaxAcceptedPerHour: envInt("COGNITIVE_ADMISSION_MAX_ACCEPTED_PER_HOUR", 60), MaxInputBytes: envInt("COGNITIVE_ADMISSION_MAX_INPUT_BYTES", 80<<10), MaxOutputBytes: envInt("COGNITIVE_ADMISSION_MAX_OUTPUT_BYTES", 16<<10)} + mux.HandleFunc("POST /v1/internal/cognitive/admission", func(w http.ResponseWriter, r *http.Request) { + var request struct { + LeaseID string `json:"lease_id"` + Request capability.AdmissionRequest `json:"request"` + } + if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 16<<10)).Decode(&request); err != nil || request.LeaseID == "" { + http.Error(w, "invalid admission request", http.StatusBadRequest) + return + } + decision, lease, err := capabilityRepository.AdmitCognitive(r.Context(), request.LeaseID, admissionPolicy, request.Request, time.Now().UnixMilli()) + if err != nil { + http.Error(w, "admission unavailable", http.StatusServiceUnavailable) + return + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(struct { + Decision capability.AdmissionDecision `json:"decision"` + Lease capability.AdmissionLease `json:"lease"` + }{decision, lease}) + }) + mux.HandleFunc("POST /v1/internal/cognitive/admission/{lease}/release", func(w http.ResponseWriter, r *http.Request) { + if err := capabilityRepository.ReleaseCognitiveAdmission(r.Context(), capability.AdmissionLease{ID: r.PathValue("lease")}, time.Now().UnixMilli()); err != nil { + http.Error(w, "admission release unavailable", http.StatusServiceUnavailable) + return + } + w.WriteHeader(http.StatusNoContent) + }) mux.HandleFunc("POST /v1/creator/sessions", creatorSessionHandler(bootstrapToken, creatorUsername, creatorPassword, sessionKey, time.Now)) mux.HandleFunc("GET /v1/creator/sessions/current", func(w http.ResponseWriter, r *http.Request) { session, ok := authenticateCreator(w, r) @@ -1159,6 +1187,14 @@ func envOr(name, fallback string) string { return fallback } +func envInt(name string, fallback int) int { + value, err := strconv.Atoi(os.Getenv(name)) + if err != nil || value < 1 { + return fallback + } + return value +} + func humanCognitiveModelInventory(candidates []capability.ResolvedCognitiveBinding, inventoryErr error) string { if inventoryErr != nil { return "Я не смогла прочитать реестр подключённых моделей. Встроенный канал Домового остаётся доступен; настройки можно проверить повторно после восстановления реестра. Ключи и пароли не выводятся в чат." diff --git a/contracts/capability/admission.go b/contracts/capability/admission.go index 1d67164..69c7b23 100644 --- a/contracts/capability/admission.go +++ b/contracts/capability/admission.go @@ -44,6 +44,11 @@ type AdmissionDecision struct { PolicyID string } +type AdmissionLease struct { + ID string + PolicyID string +} + func Admit(request AdmissionRequest, policy AdmissionPolicy, counters AdmissionCounters, nowMillis int64) AdmissionDecision { decision := AdmissionDecision{State: AdmissionAdmitted, Reason: AdmissionReasonNone, PolicyID: policy.ID} if policy.ID == "" || policy.MaxConcurrent < 1 || policy.MaxAcceptedPerHour < 1 || policy.MaxInputBytes < 1 || policy.MaxOutputBytes < 1 { diff --git a/database/migrations/033_cognitive_admission_leases.sql b/database/migrations/033_cognitive_admission_leases.sql new file mode 100644 index 0000000..ba082fe --- /dev/null +++ b/database/migrations/033_cognitive_admission_leases.sql @@ -0,0 +1,30 @@ +BEGIN; + +SET ROLE capability_role; + +CREATE TABLE IF NOT EXISTS capability.cognitive_admission_counters ( + policy_id TEXT PRIMARY KEY, + window_start_unix_millis BIGINT NOT NULL, + accepted_count INTEGER NOT NULL DEFAULT 0, + concurrent_count INTEGER NOT NULL DEFAULT 0, + updated_at_unix_millis BIGINT NOT NULL, + CHECK (policy_id <> ''), CHECK (window_start_unix_millis > 0), + CHECK (accepted_count >= 0), CHECK (concurrent_count >= 0), + CHECK (updated_at_unix_millis > 0) +); + +CREATE TABLE IF NOT EXISTS capability.cognitive_admission_leases ( + lease_id TEXT PRIMARY KEY, + policy_id TEXT NOT NULL REFERENCES capability.cognitive_admission_counters(policy_id), + state TEXT NOT NULL, + expires_at_unix_millis BIGINT NOT NULL, + created_at_unix_millis BIGINT NOT NULL, + CHECK (lease_id <> ''), CHECK (state IN ('ACTIVE','RELEASED','EXPIRED')), + CHECK (expires_at_unix_millis > 0), CHECK (created_at_unix_millis > 0) +); + +CREATE INDEX IF NOT EXISTS cognitive_admission_lease_expiry_idx + ON capability.cognitive_admission_leases(policy_id, state, expires_at_unix_millis); + +RESET ROLE; +COMMIT; diff --git a/deploy/docker-compose.prod.yml b/deploy/docker-compose.prod.yml index f5dcf94..356a638 100644 --- a/deploy/docker-compose.prod.yml +++ b/deploy/docker-compose.prod.yml @@ -80,6 +80,11 @@ services: INTERACTION_POSTGRES_CREDENTIAL_REF: credential://domovoy/interaction/db-password/v1 INTERACTION_POSTGRES_PASSWORD_FILE: /run/secrets/interaction_db_password SECRET_BACKEND_ROOT: /run/secrets + COGNITIVE_ADMISSION_POLICY_ID: creator-advisory-local-v1 + COGNITIVE_ADMISSION_MAX_CONCURRENT: ${COGNITIVE_ADMISSION_MAX_CONCURRENT:-1} + COGNITIVE_ADMISSION_MAX_ACCEPTED_PER_HOUR: ${COGNITIVE_ADMISSION_MAX_ACCEPTED_PER_HOUR:-60} + COGNITIVE_ADMISSION_MAX_INPUT_BYTES: ${COGNITIVE_ADMISSION_MAX_INPUT_BYTES:-81920} + COGNITIVE_ADMISSION_MAX_OUTPUT_BYTES: ${COGNITIVE_ADMISSION_MAX_OUTPUT_BYTES:-16384} depends_on: { postgres: { condition: service_healthy } } volumes: - assistant_core_secrets:/run/secrets:ro diff --git a/dev/ACTIVE_TASK.json b/dev/ACTIVE_TASK.json index 5694a52..1486695 100644 --- a/dev/ACTIVE_TASK.json +++ b/dev/ACTIVE_TASK.json @@ -1,5 +1,5 @@ { "task_file": "dev/tasks/DEV-000115-cognitive-admission-control.json", "task_id": "DEV-000115", - "task_revision": 1 + "task_revision": 2 } diff --git a/dev/tasks/DEV-000115-cognitive-admission-control.json b/dev/tasks/DEV-000115-cognitive-admission-control.json index aa790d9..b0d3f3e 100644 --- a/dev/tasks/DEV-000115-cognitive-admission-control.json +++ b/dev/tasks/DEV-000115-cognitive-admission-control.json @@ -1,23 +1,23 @@ { "task_id": "DEV-000115", - "task_revision": 1, + "task_revision": 2, "architecture_revision": "ADR-0018", - "base_repo_revision": "e83cf35", + "base_repo_revision": "9b26260", "risk_tier": "E", - "intent": "Add a deterministic cognitive admission policy for concurrency, accepted/hour, deadline and input/output budgets without changing provider routing or bindings.", - "requirement_ids": ["WP-203", "Bounded concurrency/rate limits", "Explainable admission decision"], - "allowed_change_scope": {"modules": ["cognitive-admission"], "paths": ["contracts/capability/**", "modules/capability/domain/**", "tests/cognitive_admission/**", "dev/ACTIVE_TASK.json", "dev/tasks/DEV-000115-cognitive-admission-control.json"]}, - "forbidden_change_scope": {"paths": ["architecture/**", "database/**", "migrations/**", "deploy/**", "security/**", "cmd/**", "runtime/**", "adapters/**", ".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 cognitive capability registry and LimitPolicyRef", "ADR-0006 bounded advisory boundary"], "may_change": ["Typed admission request/policy/decision contract"]}, + "intent": "Run bounded cognitive admission as a durable capability-owner lease with restart recovery, while preserving provider routing and explicit fallback.", + "requirement_ids": ["WP-203", "Bounded concurrency/rate limits", "Durable admission lease", "Explainable admission decision"], + "allowed_change_scope": {"modules": ["cognitive-admission", "core-runtime", "cognitive-worker", "deployment"], "paths": ["contracts/capability/**", "adapters/postgres/capability_repository.go", "database/migrations/033_cognitive_admission_leases.sql", "cmd/core-runtime/main.go", "cmd/cognitive-worker/**", "deploy/docker-compose.prod.yml", "tests/cognitive_admission/**", "dev/ACTIVE_TASK.json", "dev/tasks/DEV-000115-cognitive-admission-control.json"]}, + "forbidden_change_scope": {"paths": ["architecture/**", "security/**", "modules/authority/**", "modules/work/**", ".gitea/**"]}, + "allowed_architecture_delta": {"new_cross_module_dependencies": 0, "new_canonical_owners": 0, "new_privileges": 1, "new_third_party_dependencies": 0, "new_database_owner_roles": 1}, + "contracts": {"consume": ["Existing cognitive capability registry and LimitPolicyRef", "ADR-0006 bounded advisory boundary", "capability-owned PostgreSQL schema"], "may_change": ["Typed admission request/policy/decision and lease contract"]}, "semantic_owner": "capability-cognitive-admission", - "canonical_state": [], - "state_transitions": ["request + policy + observed counters -> admitted or typed rejection"], - "side_effects_and_privileges": ["Pure policy evaluation; no provider call, queue, persistence, binding mutation or external mutation"], - "persistence_and_migrations": ["No schema or migration changes"], - "failure_retry_restart": ["Rejected requests carry a reason; retry requires a fresh policy evaluation"], - "assumptions": ["Counters are supplied by a later durable admission implementation", "Policy does not choose or activate providers", "Deadline and budgets are request requirements"], - "acceptance_scenarios": ["Concurrency limit rejects excess request", "Accepted/hour limit rejects excess request", "Deadline and byte budget reject invalid request", "Admitted decision exposes reason and policy reference", "Simple/normal/heavy complexity remains a routing input, not authority"], - "required_tests": ["go test ./...", "go vet ./...", "go run ./cmd/devctl preflight -active dev/ACTIVE_TASK.json", "go run ./cmd/devctl check-diff -active dev/ACTIVE_TASK.json", "go run ./cmd/devctl check-architecture", "go run ./cmd/devctl/independent", "git diff --check", "admission policy tests"], + "canonical_state": ["capability.cognitive_admission_counters", "capability.cognitive_admission_leases"], + "state_transitions": ["request + policy + durable counters -> admitted lease or typed rejection", "active lease -> released/expired"], + "side_effects_and_privileges": ["Capability database role only; internal core endpoint; provider calls remain behind cognitive gateway"], + "persistence_and_migrations": ["database/migrations/033_cognitive_admission_leases.sql"], + "failure_retry_restart": ["Expired leases are recovered on next admission transaction; duplicate request IDs are idempotent; rejected requests fall back locally"], + "assumptions": ["Core Runtime remains the capability-owner admission boundary", "Policy does not choose or activate providers", "Deadline and budgets are request requirements"], + "acceptance_scenarios": ["Concurrency limit rejects excess request", "Accepted/hour limit rejects excess request", "Deadline and byte budget reject invalid request", "Admitted decision exposes reason and policy reference", "expired lease does not permanently block after restart", "cognitive worker uses admission before provider and releases lease"], + "required_tests": ["go test ./...", "go vet ./...", "go run ./cmd/devctl preflight -active dev/ACTIVE_TASK.json", "go run ./cmd/devctl check-diff -active dev/ACTIVE_TASK.json", "go run ./cmd/devctl check-architecture", "go run ./cmd/devctl/independent", "git diff --check", "compose config", "admission policy and HTTP boundary tests"], "terminal_gap_states": ["BLOCKED_BY_MISSING_CAPABILITY", "BLOCKED_BY_SPEC_GAP", "BLOCKED_BY_CONFLICT"] } -- 2.54.0