From dd9bca3ace51e1c0af50fd67dba20caad17cf391 Mon Sep 17 00:00:00 2001 From: Mikhail Date: Mon, 31 Aug 2026 10:25:16 +0300 Subject: [PATCH] fix(cognitive): trace and bound qualification flights --- cmd/cognitive-worker/model_qualification.go | 153 +++++++++++++++--- .../model_qualification_test.go | 62 +++++++ dev/ACTIVE_TASK.json | 2 +- ...E-003-restore-local-medium-invocation.json | 15 +- 4 files changed, 203 insertions(+), 29 deletions(-) diff --git a/cmd/cognitive-worker/model_qualification.go b/cmd/cognitive-worker/model_qualification.go index 0f02f04..7b71e5d 100644 --- a/cmd/cognitive-worker/model_qualification.go +++ b/cmd/cognitive-worker/model_qualification.go @@ -4,8 +4,11 @@ import ( "context" "encoding/json" "fmt" + "log" "net/http" "slices" + "sort" + "strings" "sync" "time" @@ -53,8 +56,83 @@ type modelQualificationResponse struct { Results []modelQualificationResult `json:"results"` } +const qualificationControlPlaneTimeout = 45 * time.Second + +// qualificationCoordinator bounds duplicate control-plane probes within one +// cognitive-worker process. It is deliberately not canonical state: the +// Capability owner remains the sole owner of qualification evidence and +// activation. A restart may repeat an incomplete probe, but cannot create a +// second semantic writer or bypass independent evaluation. +type qualificationCoordinator struct { + mu sync.Mutex + flights map[string]*qualificationFlight +} + +type qualificationFlight struct { + done chan struct{} + response modelQualificationResponse + err error + waiters int +} + +func (c *qualificationCoordinator) do(ctx context.Context, key string, operation func() (modelQualificationResponse, error)) (modelQualificationResponse, bool, error) { + c.mu.Lock() + if c.flights == nil { + c.flights = make(map[string]*qualificationFlight) + } + if current, exists := c.flights[key]; exists { + current.waiters++ + c.mu.Unlock() + select { + case <-current.done: + return current.response, false, current.err + case <-ctx.Done(): + return modelQualificationResponse{}, false, ctx.Err() + } + } + current := &qualificationFlight{done: make(chan struct{})} + c.flights[key] = current + c.mu.Unlock() + + current.response, current.err = operation() + close(current.done) + c.mu.Lock() + delete(c.flights, key) + c.mu.Unlock() + return current.response, true, current.err +} + +type qualificationPhaseTrace struct { + correlationID string + started time.Time +} + +func newQualificationPhaseTrace() qualificationPhaseTrace { + correlationID, err := ids.New() + if err != nil { + correlationID = ids.ID(fmt.Sprintf("qualification-%d", time.Now().UnixNano())) + } + return qualificationPhaseTrace{correlationID: string(correlationID), started: time.Now()} +} + +func (t qualificationPhaseTrace) phase(name string, fields ...string) { + log.Printf("qualification_phase correlation_id=%s phase=%s elapsed_ms=%d %s", t.correlationID, name, time.Since(t.started).Milliseconds(), strings.Join(fields, " ")) +} + +func qualificationFlightKey(provider string, modelIDs []capability.ProviderModelID) string { + ids := make([]string, 0, len(modelIDs)) + for _, modelID := range modelIDs { + ids = append(ids, string(modelID)) + } + sort.Strings(ids) + return provider + "|" + strings.Join(ids, ",") +} + func modelQualificationHandler(providers map[string]providerRegistration, owner string, observers ...qualificationObserver) http.Handler { + coordinator := &qualificationCoordinator{} return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + trace := newQualificationPhaseTrace() + trace.phase("Q0", "event=request_accepted") name := r.PathValue("provider") registration, ok := providers[name] if !ok || registration.Discovery == nil || registration.Provider == nil { @@ -72,32 +150,65 @@ func modelQualificationHandler(providers map[string]providerRegistration, owner http.Error(w, "max_output_bytes exceeds qualification bound", http.StatusBadRequest) return } - modelsFound, probe := registration.Discovery.Discover(r.Context()) - if probe.Health != capability.HealthAvailable || len(modelsFound) == 0 { - http.Error(w, "provider unavailable for qualification", http.StatusServiceUnavailable) - return + controlCtx, cancel := context.WithTimeout(r.Context(), qualificationControlPlaneTimeout) + defer cancel() + response, leader, err := coordinator.do(controlCtx, qualificationFlightKey(name, request.ModelIDs), func() (modelQualificationResponse, error) { + trace.phase("Q1", "event=binding_candidate_loaded") + modelsFound, probe := registration.Discovery.Discover(controlCtx) + if probe.Health != capability.HealthAvailable || len(modelsFound) == 0 { + return modelQualificationResponse{}, fmt.Errorf("provider unavailable for qualification") + } + if len(request.ModelIDs) > 0 { + modelsFound = slices.DeleteFunc(modelsFound, func(model models.DiscoveredModel) bool { + return !slices.Contains(request.ModelIDs, model.ProviderModelID) + }) + } + if len(modelsFound) == 0 { + return modelQualificationResponse{}, fmt.Errorf("requested model not discovered") + } + if len(modelsFound) > 16 { + modelsFound = modelsFound[:16] + } + trace.phase("Q2", fmt.Sprintf("event=provider_model_resolved model_count=%d", len(modelsFound))) + trace.phase("Q3", "event=bounded_invocation_started") + results := qualifyModels(controlCtx, registration, owner, request.Prompt, request.MaxOutputBytes, modelsFound) + trace.phase("Q4", "event=bounded_invocation_completed") + trace.phase("Q5", fmt.Sprintf("event=invocation_evidence_built result_count=%d", len(results))) + if len(observers) > 0 && observers[0] != nil { + trace.phase("Q6", "event=observation_persistence_started") + if err := persistQualificationResults(controlCtx, observers[0], registration, results); err != nil { + return modelQualificationResponse{}, fmt.Errorf("persist model qualification observation: %w", err) + } + trace.phase("Q7", "event=observation_persistence_completed") + } + // Q8-Q17 are intentionally outside this worker's authority. Capability + // owns independent evaluation and governed promotion; this trace makes + // that hand-off explicit instead of disguising it as activation. + trace.phase("Q8", "event=independent_evaluation_not_started owner=capability") + return modelQualificationResponse{Provider: registration.ProviderInstanceID, ObservedAtUnixMillis: time.Now().UnixMilli(), Results: results}, nil + }) + if !leader { + trace.phase("Q0", "event=joined_in_progress_qualification") } - if len(request.ModelIDs) > 0 { - modelsFound = slices.DeleteFunc(modelsFound, func(model models.DiscoveredModel) bool { - return !slices.Contains(request.ModelIDs, model.ProviderModelID) - }) - } - if len(modelsFound) == 0 { - http.Error(w, "requested model not discovered", http.StatusNotFound) - return - } - if len(modelsFound) > 16 { - modelsFound = modelsFound[:16] - } - results := qualifyModels(r.Context(), registration, owner, request.Prompt, request.MaxOutputBytes, modelsFound) - if len(observers) > 0 && observers[0] != nil { - if err := persistQualificationResults(r.Context(), observers[0], registration, results); err != nil { - http.Error(w, "persist model qualification observation", http.StatusBadGateway) + if err != nil { + if controlCtx.Err() != nil { + http.Error(w, "qualification timeout", http.StatusGatewayTimeout) return } + if strings.Contains(err.Error(), "requested model not discovered") { + http.Error(w, "requested model not discovered", http.StatusNotFound) + return + } + if strings.Contains(err.Error(), "provider unavailable") { + http.Error(w, "provider unavailable for qualification", http.StatusServiceUnavailable) + return + } + http.Error(w, "persist model qualification observation", http.StatusBadGateway) + return } + trace.phase("Q17", "event=qualification_response_returned activation=not_attempted") w.Header().Set("Content-Type", "application/json") - _ = jsonEncode(w, modelQualificationResponse{Provider: registration.ProviderInstanceID, ObservedAtUnixMillis: time.Now().UnixMilli(), Results: results}) + _ = jsonEncode(w, response) }) } diff --git a/cmd/cognitive-worker/model_qualification_test.go b/cmd/cognitive-worker/model_qualification_test.go index b4e05e8..f736b78 100644 --- a/cmd/cognitive-worker/model_qualification_test.go +++ b/cmd/cognitive-worker/model_qualification_test.go @@ -6,6 +6,7 @@ import ( "net/http/httptest" "strings" "testing" + "time" "git.cin.su/m/Assistant/adapters/models" "git.cin.su/m/Assistant/contracts/capability" @@ -17,6 +18,7 @@ type qualificationProvider struct{} func (qualificationProvider) Probe(context.Context) models.ProviderProbe { return models.ProviderProbe{Health: capability.HealthAvailable} } + func (qualificationProvider) Discover(context.Context) ([]models.DiscoveredModel, models.ProviderProbe) { return []models.DiscoveredModel{{ProviderModelID: "model-a"}, {ProviderModelID: "model-b"}}, models.ProviderProbe{Health: capability.HealthAvailable} } @@ -83,3 +85,63 @@ func TestModelQualificationPersistsReachabilityWithoutQuality(t *testing.T) { t.Fatalf("invalid qualification observation: %+v err=%v", observer.observations[0], err) } } + +func TestModelQualificationDeduplicatesSameProviderModelFlight(t *testing.T) { + coordinator := &qualificationCoordinator{} + started := make(chan struct{}) + release := make(chan struct{}) + firstDone := make(chan struct{}) + secondDone := make(chan struct{}) + result := modelQualificationResponse{Provider: "medium"} + go func() { + _, _, err := coordinator.do(context.Background(), "medium|model-a", func() (modelQualificationResponse, error) { + close(started) + <-release + return result, nil + }) + if err != nil { + t.Errorf("leader qualification failed: %v", err) + } + close(firstDone) + }() + select { + case <-started: + case <-time.After(time.Second): + t.Fatal("qualification invocation did not start") + } + go func() { + actual, leader, err := coordinator.do(context.Background(), "medium|model-a", func() (modelQualificationResponse, error) { + t.Error("duplicate qualification operation ran") + return modelQualificationResponse{}, nil + }) + if err != nil || leader || actual.Provider != result.Provider { + t.Errorf("joined qualification response=%+v leader=%t err=%v", actual, leader, err) + } + close(secondDone) + }() + deadline := time.Now().Add(time.Second) + for { + coordinator.mu.Lock() + flight := coordinator.flights["medium|model-a"] + waiters := 0 + if flight != nil { + waiters = flight.waiters + } + coordinator.mu.Unlock() + if waiters == 1 { + break + } + if time.Now().After(deadline) { + t.Fatal("second qualification did not join the in-progress flight") + } + time.Sleep(time.Millisecond) + } + close(release) + for _, done := range []chan struct{}{firstDone, secondDone} { + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("qualification flight did not complete") + } + } +} diff --git a/dev/ACTIVE_TASK.json b/dev/ACTIVE_TASK.json index 355aafd..ffefd21 100644 --- a/dev/ACTIVE_TASK.json +++ b/dev/ACTIVE_TASK.json @@ -1 +1 @@ -{"task_file":"dev/tasks/GEN2-COGNITIVE-003-restore-local-medium-invocation.json","task_id":"GEN2-COGNITIVE-003","task_revision":3} +{"task_file":"dev/tasks/GEN2-COGNITIVE-003-restore-local-medium-invocation.json","task_id":"GEN2-COGNITIVE-003","task_revision":4} diff --git a/dev/tasks/GEN2-COGNITIVE-003-restore-local-medium-invocation.json b/dev/tasks/GEN2-COGNITIVE-003-restore-local-medium-invocation.json index a4bd06c..34cc7a4 100644 --- a/dev/tasks/GEN2-COGNITIVE-003-restore-local-medium-invocation.json +++ b/dev/tasks/GEN2-COGNITIVE-003-restore-local-medium-invocation.json @@ -1,17 +1,18 @@ { "task_id": "GEN2-COGNITIVE-003", - "task_revision": 3, + "task_revision": 4, "architecture_revision": "ARCH-GEN2-000", "base_repo_revision": "gen2-integration", "risk_tier": "E", - "intent": "Restore bounded OpenAI-compatible Local MEDIUM invocation without changing canonical ownership or Core semantics.", - "requirement_ids": ["REQ-GEN2-EVOLVE-004-LOCAL-MEDIUM-INVOCATION"], + "intent": "Restore and phase-trace bounded Local MEDIUM qualification and invocation without changing Capability ownership, Core household budget, or activation authority.", + "requirement_ids": ["REQ-GEN2-EVOLVE-004-LOCAL-MEDIUM-INVOCATION", "REQ-GEN2-COGNITIVE-003-QUALIFICATION-CONTROL-PLANE"], "allowed_change_scope": { "modules": ["cognitive-worker provider invocation", "existing local OpenAI-compatible adapter", "targeted evidence"], "paths": [ "cmd/cognitive-worker/main.go", "cmd/cognitive-worker/advisory.go", "cmd/cognitive-worker/gateway_registry.go", + "cmd/cognitive-worker/model_qualification.go", "cmd/cognitive-worker/*_test.go", "adapters/models/openai_compatible_local.go", "adapters/models/openai_compatible_local_test.go", @@ -66,13 +67,13 @@ "may_change": ["provider request normalization and typed immediate failure mapping only"] }, "semantic_owner": "Cognitive gateway remains the sole owner of provider invocation and route selection.", - "canonical_state": ["discovery is not invocation qualification", "optional provider parameters are capability-gated", "immediate provider rejection is not reported as timeout"], - "state_transitions": ["binding selection -> normalized request -> provider invocation -> typed result"], + "canonical_state": ["discovery is not invocation qualification", "qualification evidence is not activation", "optional provider parameters are capability-gated", "immediate provider rejection is not reported as timeout"], + "state_transitions": ["binding selection -> normalized request -> provider invocation -> typed result", "discovery -> bounded qualification evidence -> independent capability evaluation -> governed canary/activation"], "side_effects_and_privileges": ["no secrets or authorization headers in evidence", "no direct database access", "no provider calls outside the existing gateway boundary"], "persistence_and_migrations": ["no schema change"], - "failure_retry_restart": ["bounded invocation; no timeout inflation; no infinite retry; preserve fail-closed/fallback semantics"], + "failure_retry_restart": ["household invocation remains bounded by its existing 12-second Core budget", "qualification/revalidation is a separately bounded control-plane operation", "single-flight qualification per provider/model avoids duplicate work", "no timeout inflation; no infinite retry; preserve fail-closed/fallback semantics"], "assumptions": ["Local MEDIUM gateway remains the existing provider endpoint", "candidate slot B is available for isolated verification", "cmd/core-runtime/learning_test.go is a preserved carry-forward regression from the preceding runtime task"], - "acceptance_scenarios": ["minimal invocation completes", "ordinary and bounded-context invocation complete", "immediate provider 4xx maps to typed rejection without timeout", "qualification requires bounded invocation evidence"], + "acceptance_scenarios": ["safe phase trace emits correlation IDs and durations without prompts, answers, credentials, secrets, or user memory", "minimal invocation completes", "ordinary and bounded-context invocation complete", "immediate provider 4xx maps to typed rejection without timeout", "qualification requires bounded invocation evidence", "duplicate qualification is bounded and deduplicated", "fresh validated binding is not synchronously requalified on an ordinary household request"], "required_tests": ["targeted go test", "targeted go vet", "preflight", "check-diff", "independent", "architecture checks", "git diff --check"], "terminal_gap_states": ["BLOCKED_BY_ARCHITECTURE_GAP", "BLOCKED_BY_SPEC_GAP", "BLOCKED_BY_MISSING_CAPABILITY", "REQUIRES_MIGRATION_DECISION", "BLOCKED_BY_DEPLOYMENT_CONTROL"] }