Compare commits

..
Author SHA1 Message Date
m 7978b495d1 feat(reflection): expose worker runtime status
CI / test (push) Successful in 48s
CI / test (pull_request) Successful in 46s
2026-08-23 19:42:07 +03:00
m d73d4fc118 Merge pull request 'feat(capability): consume qualification evidence in selection' (#220) from codex/dev-000243-qualification-selection-evidence into main
CI / test (push) Successful in 1m8s
feat(capability): consume qualification evidence in selection
2026-08-23 16:30:23 +00:00
m c81e2269e1 feat(capability): consume qualification evidence in selection
CI / test (push) Successful in 1m16s
CI / test (pull_request) Successful in 1m26s
2026-08-23 19:28:13 +03:00
m 47327cc6d8 Merge pull request 'feat(capability): persist model qualification observations' (#219) from codex/dev-000242-durable-qualification-observations into main
CI / test (push) Successful in 1m16s
feat(capability): persist model qualification observations
2026-08-23 16:22:58 +00:00
10 changed files with 219 additions and 2 deletions
@@ -505,6 +505,14 @@ func (r *CapabilityRepository) loadCognitiveEvidence(ctx context.Context, value
if err == nil {
value.Performance = &performance
}
var qualification capability.ModelQualificationObservation
err = r.pool.QueryRow(ctx, `SELECT observation_id,provider_instance_id,provider_model_id,status,finish_reason,latency_millis,output_observed,quality_assessed,category,observed_at_unix_millis FROM capability.cognitive_model_qualification_observations WHERE provider_instance_id=$1 AND provider_model_id=$2 ORDER BY observed_at_unix_millis DESC LIMIT 1`, value.Provider.ID, value.Model.ProviderModelID).Scan(&qualification.ID, &qualification.ProviderInstanceID, &qualification.ProviderModelID, &qualification.Status, &qualification.FinishReason, &qualification.LatencyMillis, &qualification.OutputObserved, &qualification.QualityAssessed, &qualification.Category, &qualification.ObservedAtUnixMillis)
if err != nil && !errors.Is(err, pgx.ErrNoRows) {
return err
}
if err == nil {
value.Qualification = &qualification
}
if profileID != "" {
var taskEvaluation capability.TaskEvaluation
err = r.pool.QueryRow(ctx, `SELECT evaluation_id,task_profile_id,binding_id,binding_revision,capability_id,result,quality_score_millis,reliability_score_millis,latency_score_millis,overall_score_millis,critical_failure_millis,benchmark_version,artifact_id,observed_at_unix_millis FROM capability.cognitive_task_evaluations WHERE task_profile_id=$1 AND binding_id=$2 AND binding_revision=$3 ORDER BY observed_at_unix_millis DESC LIMIT 1`, profileID, value.Binding.ID, value.Binding.Revision).Scan(&taskEvaluation.ID, &taskEvaluation.ProfileID, &taskEvaluation.BindingID, &taskEvaluation.BindingRevision, &taskEvaluation.CapabilityID, &taskEvaluation.Result, &taskEvaluation.QualityScoreMillis, &taskEvaluation.ReliabilityScoreMillis, &taskEvaluation.LatencyScoreMillis, &taskEvaluation.OverallScoreMillis, &taskEvaluation.CriticalFailureMillis, &taskEvaluation.BenchmarkVersion, &taskEvaluation.ArtifactID, &taskEvaluation.ObservedAtUnixMillis)
+17
View File
@@ -4,6 +4,8 @@ import (
"context"
"flag"
"fmt"
"log"
"net/http"
"os"
"time"
@@ -29,9 +31,21 @@ func main() {
panic(err)
}
defer repository.Close()
status := newWorkerStatus(*workerID, *once)
listenAddr := os.Getenv("REFLECTION_LISTEN_ADDR")
if listenAddr == "" {
listenAddr = ":8092"
}
go func() {
if err := http.ListenAndServe(listenAddr, status.handler()); err != nil {
log.Printf("reflection status server stopped: %v", err)
}
}()
for {
now := time.Now().UnixMilli()
status.cycleStarted(time.Now())
if _, err := repository.RecoverReflectionClaims(ctx, now); err != nil {
status.cycleFailed(time.Now(), err)
panic(err)
}
candidate, claimErr := repository.ClaimReflectionCandidate(ctx, *workerID, now, lease.Milliseconds())
@@ -39,16 +53,19 @@ func main() {
// This worker only closes the durable evaluation loop. Any cognitive
// proposal remains a candidate and must be admitted by a separate governed flow.
if err := repository.ResolveReflectionCandidate(ctx, candidate.ID, *workerID, time.Now().UnixMilli()); err != nil {
status.cycleFailed(time.Now(), err)
panic(err)
}
fmt.Printf("resolved reflection candidate %s\n", candidate.ID)
} else if *once {
fmt.Println("no reflection candidate")
}
status.cycleFinished(time.Now(), map[bool]string{true: "candidate-resolved", false: "idle"}[claimErr == nil])
if *once {
return
}
if err := repository.WaitForReflectionCandidate(ctx, *pollInterval); err != nil {
status.cycleFailed(time.Now(), err)
panic(err)
}
}
+74
View File
@@ -0,0 +1,74 @@
package main
import (
"encoding/json"
"net/http"
"sync"
"time"
"git.cin.su/m/Assistant/runtime/health"
)
type workerStatus struct {
mu sync.RWMutex
workerID string
once bool
startedAt time.Time
lastCycleAt time.Time
lastCompleted time.Time
lastResult string
lastError string
}
func newWorkerStatus(workerID string, once bool) *workerStatus {
return &workerStatus{workerID: workerID, once: once, startedAt: time.Now().UTC()}
}
func (s *workerStatus) cycleStarted(at time.Time) {
s.mu.Lock()
defer s.mu.Unlock()
s.lastCycleAt = at.UTC()
s.lastResult = "running"
s.lastError = ""
}
func (s *workerStatus) cycleFinished(at time.Time, result string) {
s.mu.Lock()
defer s.mu.Unlock()
s.lastCompleted = at.UTC()
s.lastResult = result
s.lastError = ""
}
func (s *workerStatus) cycleFailed(at time.Time, err error) {
s.mu.Lock()
defer s.mu.Unlock()
s.lastCompleted = at.UTC()
s.lastResult = "failed"
s.lastError = err.Error()
}
func (s *workerStatus) handler() http.Handler {
mux := http.NewServeMux()
mux.Handle("/healthz", health.Handler{Runtime: "reflection-worker", SemanticGeneration: 1, ArchitectureGeneration: 1})
mux.HandleFunc("/status", func(w http.ResponseWriter, _ *http.Request) {
s.mu.RLock()
response := struct {
Runtime string `json:"runtime"`
WorkerID string `json:"worker_id"`
Mode string `json:"mode"`
StartedAt time.Time `json:"started_at"`
LastCycleAt time.Time `json:"last_cycle_at,omitempty"`
LastCompleted time.Time `json:"last_completed_at,omitempty"`
LastResult string `json:"last_result"`
LastError string `json:"last_error,omitempty"`
}{Runtime: "reflection-worker", WorkerID: s.workerID, Mode: "durable-reflection-loop", StartedAt: s.startedAt, LastCycleAt: s.lastCycleAt, LastCompleted: s.lastCompleted, LastResult: s.lastResult, LastError: s.lastError}
if s.once {
response.Mode = "durable-reflection-once"
}
s.mu.RUnlock()
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode(response)
})
return mux
}
+35
View File
@@ -0,0 +1,35 @@
package main
import (
"net/http"
"net/http/httptest"
"strings"
"testing"
"time"
)
func TestWorkerStatusReportsCycle(t *testing.T) {
status := newWorkerStatus("test-worker", false)
status.cycleStarted(time.Date(2026, 8, 23, 10, 0, 0, 0, time.UTC))
status.cycleFinished(time.Date(2026, 8, 23, 10, 0, 1, 0, time.UTC), "idle")
recorder := httptest.NewRecorder()
status.handler().ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/status", nil))
if recorder.Code != http.StatusOK {
t.Fatalf("status code = %d, want 200", recorder.Code)
}
body := recorder.Body.String()
for _, expected := range []string{"\"runtime\":\"reflection-worker\"", "\"worker_id\":\"test-worker\"", "\"last_result\":\"idle\""} {
if !strings.Contains(body, expected) {
t.Fatalf("status body missing %q: %s", expected, body)
}
}
}
func TestWorkerHealthIsAvailable(t *testing.T) {
recorder := httptest.NewRecorder()
newWorkerStatus("test-worker", true).handler().ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, "/healthz", nil))
if recorder.Code != http.StatusOK {
t.Fatalf("health status code = %d, want 200", recorder.Code)
}
}
@@ -172,6 +172,7 @@ type ResolvedCognitiveBinding struct {
Evaluation *EvaluatedCapability
Performance *PerformanceEvidence
TaskEvaluation *TaskEvaluation
Qualification *ModelQualificationObservation
Rating *ModelRating
}
+2 -2
View File
@@ -1,5 +1,5 @@
{
"task_file": "dev/tasks/DEV-000242-durable-qualification-observations.json",
"task_id": "DEV-000242",
"task_file": "dev/tasks/DEV-000244-reflection-worker-observability.json",
"task_id": "DEV-000244",
"task_revision": 1
}
@@ -0,0 +1,23 @@
{
"task_id": "DEV-000243",
"task_revision": 1,
"architecture_revision": "ADR-0018",
"base_repo_revision": "47327cc6d853da9084c96e3bf48502266706ca57",
"risk_tier": "E",
"intent": "Make Capability selection consume the latest durable qualification observation for fresh unavailability exclusion and transparent latency rating, without granting quality or activation.",
"requirement_ids": ["qualification selection loop", "hardware-neutral routing", "quality separated from reachability"],
"allowed_change_scope": {"modules": ["capability", "adapters/postgres"], "paths": ["adapters/postgres/capability_repository.go", "contracts/capability/cognitive_registry.go", "modules/capability/domain/cognitive_registry.go", "modules/capability/domain/model_rating_test.go", "dev/ACTIVE_TASK.json", "dev/tasks/DEV-000243-qualification-selection-evidence.json"]},
"forbidden_change_scope": {"paths": ["architecture/**", "security/**", "database/**", "deploy/**", "LLM-local/**", "modules/work/secrets/**", "contracts/work/**"]},
"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": ["durable ModelQualificationObservation"], "may_change": ["resolved binding evidence projection"]},
"semantic_owner": "Capability selection and rating",
"canonical_state": ["qualification observation remains evidence; no activation mutation"],
"state_transitions": ["fresh UNAVAILABLE observation -> candidate excluded", "stale UNAVAILABLE observation -> discovery state remains authoritative", "REACHABLE observation -> latency evidence only"],
"side_effects_and_privileges": ["read-only projection and deterministic selection"],
"persistence_and_migrations": ["uses migration 045 from DEV-000242"],
"failure_retry_restart": ["missing observation preserves existing behavior"],
"assumptions": ["provider/model observed timestamps define freshness boundary"],
"acceptance_scenarios": ["fresh unavailable candidate is not selected", "stale unavailable candidate is not permanently poisoned", "quality remains task-evaluation controlled"],
"required_tests": ["go test ./modules/capability/... ./adapters/postgres/...", "go vet ./modules/capability/... ./adapters/postgres/...", "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", "git diff --check", "independent", "CI", "production health smoke"],
"terminal_gap_states": ["BLOCKED_BY_MISSING_CAPABILITY", "BLOCKED_BY_SPEC_GAP"]
}
@@ -0,0 +1,26 @@
{
"task_id": "DEV-000244",
"task_revision": 1,
"architecture_revision": "ADR-0018",
"base_repo_revision": "c81e2269e157c1a41b567d79d93bbbe49dd4b134",
"risk_tier": "E",
"intent": "Expose bounded health and status evidence for the durable reflection worker without changing reflection semantics or authority.",
"requirement_ids": ["learning reflection loop", "runtime observability", "fast operational verification"],
"allowed_change_scope": {
"modules": ["epistemics"],
"paths": ["cmd/reflection-worker/**", "dev/ACTIVE_TASK.json", "dev/tasks/DEV-000244-reflection-worker-observability.json"]
},
"forbidden_change_scope": {"paths": ["architecture/**", "security/**", "database/**", "LLM-local/**", "modules/work/secrets/**", "contracts/**"]},
"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 reflection repository port", "runtime health conventions"], "may_change": ["worker operational HTTP surface"]},
"semantic_owner": "Epistemics reflection worker runtime",
"canonical_state": ["none; status is derived operational evidence"],
"state_transitions": ["cycle start/complete/failure are reported without changing candidate lifecycle rules"],
"side_effects_and_privileges": ["read-only health/status endpoint; existing repository writes unchanged"],
"persistence_and_migrations": ["none"],
"failure_retry_restart": ["status resets on process restart; durable candidate state remains authoritative"],
"assumptions": ["internal production network is sufficient for worker status checks"],
"acceptance_scenarios": ["health endpoint returns 200", "status exposes loop mode and latest cycle", "reflection loop still resolves candidates"],
"required_tests": ["go test ./cmd/reflection-worker", "go vet ./cmd/reflection-worker", "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", "git diff --check", "independent", "CI", "production health smoke"],
"terminal_gap_states": ["BLOCKED_BY_MISSING_CAPABILITY", "BLOCKED_BY_SPEC_GAP"]
}
@@ -148,6 +148,9 @@ func SelectCognitiveBinding(requirement capability.CognitiveRequirement, candida
if candidate.Evaluation != nil && candidate.Evaluation.Result == capability.EvaluationUnsupported {
continue
}
if candidate.Qualification != nil && candidate.Qualification.Status == capability.ModelQualificationUnavailable && candidate.Qualification.ObservedAtUnixMillis >= maxObservedAt(candidate.Provider.ObservedAtUnixMillis, candidate.Model.ObservedAtUnixMillis) {
continue
}
if requirement.TaskProfileID != "" && (candidate.TaskEvaluation == nil || candidate.TaskEvaluation.ProfileID != requirement.TaskProfileID || candidate.TaskEvaluation.Result != capability.EvaluationSupported) {
continue
}
@@ -195,10 +198,14 @@ func RateCognitiveBinding(candidate capability.ResolvedCognitiveBinding) capabil
latency := int64(500)
if candidate.Performance != nil && candidate.Performance.LatencyMicros != nil {
latency = latencyScore(*candidate.Performance.LatencyMicros)
} else if candidate.Qualification != nil && candidate.Qualification.Status == capability.ModelQualificationReachable {
latency = latencyScore(candidate.Qualification.LatencyMillis * 1000)
}
evidence := int64(700)
if candidate.Performance != nil {
evidence = 1000
} else if candidate.Qualification != nil && candidate.Qualification.Status == capability.ModelQualificationReachable {
evidence = 750
}
overall := (quality*55 + reliability*25 + latency*15 + evidence*5) / 100
category := capability.ModelRatingBalanced
@@ -210,6 +217,13 @@ func RateCognitiveBinding(candidate capability.ResolvedCognitiveBinding) capabil
return capability.ModelRating{Category: category, OverallScoreMillis: overall, QualityScoreMillis: quality, ReliabilityScoreMillis: reliability, LatencyScoreMillis: latency, EvidenceScoreMillis: evidence}
}
func maxObservedAt(left, right int64) int64 {
if left > right {
return left
}
return right
}
func clampMillis(value int64) int64 {
if value < 0 {
return 0
@@ -38,3 +38,22 @@ func TestRateCognitiveBindingClassifiesSlowEvidenceAsBalanced(t *testing.T) {
t.Fatalf("rating=%+v", rating)
}
}
func TestSelectionExcludesFreshUnavailableQualificationButNotStaleEvidence(t *testing.T) {
quality, critical := int64(900), int64(0)
base := capability.ResolvedCognitiveBinding{
Binding: capability.CognitiveBinding{ID: "binding-qualification", Revision: 1, CapabilityID: "creator-advisory", OwnerPrincipalID: "owner", ProviderInstanceID: "provider", ProviderRevision: 1, ModelDescriptorID: "model", ModelRevision: 1, DataPolicy: capability.DataPolicyLocalOnly, Qualification: capability.QualificationValidated, Activation: capability.ActivationActive, Health: capability.HealthAvailable, LimitPolicyRef: "limits"},
Provider: capability.ProviderInstance{ID: "provider", Revision: 1, Kind: "local", EndpointRef: "endpoint", Enabled: true, Qualification: capability.QualificationValidated, Health: capability.HealthAvailable, ObservedAtUnixMillis: 20},
Model: capability.ModelDescriptor{ID: "model", Revision: 1, ProviderInstanceID: "provider", ProviderModelID: "model", MaxContextBytes: 1000, Modalities: []capability.ModelModality{capability.ModalityText}, Capabilities: []capability.ModelCapability{capability.CapabilityAdvisory}, Qualification: capability.QualificationValidated, Health: capability.HealthAvailable, ObservedAtUnixMillis: 20},
Evaluation: &capability.EvaluatedCapability{ID: "evaluation", BindingID: "binding-qualification", BindingRevision: 1, CapabilityID: "creator-advisory", Result: capability.EvaluationSupported, QualityScoreMillis: &quality, CriticalFailureMillis: &critical},
Qualification: &capability.ModelQualificationObservation{ID: "observation", ProviderInstanceID: "provider", ProviderModelID: "model", Status: capability.ModelQualificationUnavailable, FinishReason: "TIMEOUT", ObservedAtUnixMillis: 20, Category: capability.ModelRatingUnrated},
}
requirement := capability.CognitiveRequirement{CapabilityID: "creator-advisory", OwnerPrincipalID: "owner", DataPolicy: capability.DataPolicyLocalOnly, RequiredModalities: []capability.ModelModality{capability.ModalityText}, RequiredCapabilities: []capability.ModelCapability{capability.CapabilityAdvisory}, MaxOutputBytes: 10}
if _, err := SelectCognitiveBinding(requirement, []capability.ResolvedCognitiveBinding{base}); err == nil {
t.Fatal("fresh unavailable qualification should exclude candidate")
}
base.Qualification.ObservedAtUnixMillis = 19
if _, err := SelectCognitiveBinding(requirement, []capability.ResolvedCognitiveBinding{base}); err != nil {
t.Fatalf("stale unavailable qualification should not exclude candidate: %v", err)
}
}