diff --git a/adapters/postgres/epistemics_repository.go b/adapters/postgres/epistemics_repository.go index 3f1a9bf..2f9c0fd 100644 --- a/adapters/postgres/epistemics_repository.go +++ b/adapters/postgres/epistemics_repository.go @@ -6,6 +6,7 @@ import ( "time" "git.cin.su/m/Assistant/contracts/epistemics" + "git.cin.su/m/Assistant/platform/telemetry" "github.com/jackc/pgx/v5/pgxpool" ) @@ -36,6 +37,11 @@ func (r *EpistemicsRepository) Ping(ctx context.Context) error { // TruthRuntimeStatus is a read-only projection of durable truth bundles. // It does not promote claims, revise beliefs, or mutate epistemic state. func (r *EpistemicsRepository) TruthRuntimeStatus(ctx context.Context) (TruthRuntimeStatus, error) { + started := time.Now() + outcome := "SUCCESS" + defer func() { + telemetry.RecordCompatibility("truth-bundles-status-reader", "epistemics-repository", "read", "research-status", true, false, true, outcome, "truth-status", "epistemics", started) + }() status := TruthRuntimeStatus{ ClaimsByVerification: map[string]int64{}, BeliefsByState: map[string]int64{}, @@ -43,6 +49,7 @@ func (r *EpistemicsRepository) TruthRuntimeStatus(ctx context.Context) (TruthRun ObservedAtUnixMillis: time.Now().UnixMilli(), } if err := r.pool.QueryRow(ctx, `SELECT count(*) FROM epistemics.truth_bundles`).Scan(&status.BundleCount); err != nil { + outcome = "FAILURE" return TruthRuntimeStatus{}, err } queries := []struct { @@ -56,18 +63,21 @@ func (r *EpistemicsRepository) TruthRuntimeStatus(ctx context.Context) (TruthRun for _, item := range queries { rows, err := r.pool.Query(ctx, item.query) if err != nil { + outcome = "FAILURE" return TruthRuntimeStatus{}, err } for rows.Next() { var state string var count int64 if err := rows.Scan(&state, &count); err != nil { + outcome = "FAILURE" rows.Close() return TruthRuntimeStatus{}, err } item.destination[state] = count } if err := rows.Err(); err != nil { + outcome = "FAILURE" rows.Close() return TruthRuntimeStatus{}, err } @@ -218,7 +228,13 @@ func (r *EpistemicsRepository) SaveReflectionCandidate(ctx context.Context, valu } func (r *EpistemicsRepository) SaveTruthBundle(ctx context.Context, value epistemics.TruthBundle, recordedAtUnixMillis int64) error { + started := time.Now() + outcome := "SUCCESS" + defer func() { + telemetry.RecordCompatibility("truth-bundles-writer", "epistemics-repository", "write", "persisted-research", true, false, true, outcome, string(value.Claim.ID), "epistemics", started) + }() if value.Source.ID == "" || value.Snapshot.ID == "" || value.Excerpt.ID == "" || value.Claim.ID == "" || value.Belief.ID == "" || value.Citation.ID == "" || recordedAtUnixMillis < 1 { + outcome = "INVALID" return fmt.Errorf("invalid truth bundle") } _, err := r.pool.Exec(ctx, `INSERT INTO epistemics.truth_bundles ( @@ -232,5 +248,8 @@ func (r *EpistemicsRepository) SaveTruthBundle(ctx context.Context, value episte value.Excerpt.ID, value.Excerpt.SnapshotID, value.Excerpt.Text, value.Excerpt.Start, value.Excerpt.End, value.Observation.ID, value.Observation.ObservedAt, value.Claim.ID, value.Claim.Statement, value.Claim.ObservationID, value.Claim.CitationID, value.Claim.Verification, value.Belief.ID, value.Belief.ClaimID, value.Belief.State, value.Citation.ID, value.Citation.ExcerptID, value.Citation.Evidence, value.Citation.Coverage, value.Citation.Conflict, value.Citation.ResolvedAtUnixMillis, recordedAtUnixMillis) + if err != nil { + outcome = "FAILURE" + } return err } diff --git a/cmd/core-runtime/learning.go b/cmd/core-runtime/learning.go index d53398b..25e762b 100644 --- a/cmd/core-runtime/learning.go +++ b/cmd/core-runtime/learning.go @@ -13,6 +13,7 @@ import ( "git.cin.su/m/Assistant/contracts/interaction" capabilitydomain "git.cin.su/m/Assistant/modules/capability/domain" epistemicsdomain "git.cin.su/m/Assistant/modules/epistemics/domain" + "git.cin.su/m/Assistant/platform/telemetry" ) const reminderResponseSkillID = capability.SkillProfileID("skill-obvious-reminder-response-v1") @@ -160,6 +161,7 @@ func (r learningRecorder) recordFastPathDecline(ctx context.Context, input fastP if r.epistemics == nil || input.Intent.ID == "" || input.Utterance.ID == "" { return } + telemetry.RecordCompatibility("fast-path-v1", "core-runtime", "decision", "decline-or-escalate", true, true, false, "OBSERVED", string(input.Utterance.CorrelationID), string(input.Utterance.PrincipalID), input.At) experience, err := epistemicsdomain.FastPathDeclineExperience( epistemics.OriginalIntentRef(input.Intent.ID), epistemics.SourceUtteranceRef(input.Utterance.ID), @@ -273,9 +275,11 @@ func (r learningRecorder) updateReminderSkill(ctx context.Context, experience ep } profile, err := r.capability.LoadSkillProfile(ctx, reminderResponseSkillID) if err != nil { + telemetry.RecordCompatibility("skill-profile-v1", "core-runtime", "read", "fast-path-learning", true, false, true, "FAILURE", string(experience.OriginalIntentID), string(experience.SourceUtteranceID), time.UnixMilli(experience.RecordedAtUnixMillis)) log.Printf("learning skill profile unavailable: %v", err) return } + telemetry.RecordCompatibility("skill-profile-v1", "core-runtime", "read", "fast-path-learning", true, false, true, "SUCCESS", string(experience.OriginalIntentID), string(experience.SourceUtteranceID), time.UnixMilli(experience.RecordedAtUnixMillis)) next, err := capabilitydomain.ObserveSkillOutcome(profile, capability.SkillOutcomeObservation{ ExperienceRef: capability.SkillEvidenceRef("epistemics:experience:" + string(experience.ID)), Outcome: outcome, @@ -287,9 +291,11 @@ func (r learningRecorder) updateReminderSkill(ctx context.Context, experience ep return } if err := r.capability.SaveSkillProfile(ctx, next); err != nil { + telemetry.RecordCompatibility("skill-profile-v1", "core-runtime", "write", "fast-path-learning", true, false, true, "FAILURE", string(experience.OriginalIntentID), string(experience.SourceUtteranceID), time.UnixMilli(experience.RecordedAtUnixMillis)) log.Printf("learning skill profile save unavailable: %v", err) return } + telemetry.RecordCompatibility("skill-profile-v1", "core-runtime", "write", "fast-path-learning", true, false, true, "SUCCESS", string(experience.OriginalIntentID), string(experience.SourceUtteranceID), time.UnixMilli(experience.RecordedAtUnixMillis)) } func outcomeForHumanResponse(response continuity.HumanResponse) (epistemics.OutcomeClass, int64) { diff --git a/cmd/core-runtime/main.go b/cmd/core-runtime/main.go index 5e006ce..8f371e5 100644 --- a/cmd/core-runtime/main.go +++ b/cmd/core-runtime/main.go @@ -24,6 +24,7 @@ import ( workdomain "git.cin.su/m/Assistant/modules/work/domain" "git.cin.su/m/Assistant/platform/ids" platformsecrets "git.cin.su/m/Assistant/platform/secrets" + "git.cin.su/m/Assistant/platform/telemetry" "git.cin.su/m/Assistant/runtime/health" "github.com/jackc/pgx/v5" "io" @@ -65,6 +66,8 @@ func main() { if creatorUsername == "" { log.Fatal("CREATOR_USERNAME is required") } + telemetry.RecordCompatibility("non-bootstrap-env-consumers", "core-runtime", "read", "runtime-startup-configuration", true, true, true, "OBSERVED", "startup", "core-runtime", time.Now()) + telemetry.RecordCompatibility("migration-compatibility-switches", "core-runtime", "observe", "runtime-startup-switch-scan", true, false, false, "NOT_OBSERVED", "startup", "core-runtime", time.Now()) repository, err := postgres.OpenSituation(context.Background(), postgres.ContinuityConfig{Host: os.Getenv("POSTGRES_HOST"), Port: os.Getenv("POSTGRES_PORT"), Database: os.Getenv("POSTGRES_DATABASE"), User: os.Getenv("POSTGRES_USER"), Password: os.Getenv("POSTGRES_PASSWORD")}) if err != nil { log.Fatal(err) @@ -553,9 +556,13 @@ func main() { storedFocus, err := repository.ActiveProjectFocus(r.Context(), session.ActorID, session.PrincipalID) if err == nil { previousFocus = &storedFocus + telemetry.RecordCompatibility("active-project-focus-read", "core-runtime", "read", "chat-focus-load", true, false, true, "SUCCESS", utterance.CorrelationID, string(session.PrincipalID), now) } else if !errors.Is(err, pgx.ErrNoRows) { + telemetry.RecordCompatibility("active-project-focus-read", "core-runtime", "read", "chat-focus-load", true, false, true, "FAILURE", utterance.CorrelationID, string(session.PrincipalID), now) http.Error(w, "project focus unavailable", http.StatusServiceUnavailable) return + } else { + telemetry.RecordCompatibility("active-project-focus-read", "core-runtime", "read", "chat-focus-load", true, false, true, "EMPTY", utterance.CorrelationID, string(session.PrincipalID), now) } var effectiveTaskMessage = interactiondomain.StripAssistantVocative(utterance.Content) creativeUserRequest := interactiondomain.IsCreativeRequest(utterance.Content) @@ -582,6 +589,9 @@ func main() { } conversationHistory = append(conversationHistory, recentUserHistory...) gen2Context := interactiondomain.BuildScopedGen2Context(conversationHistory, string(session.PrincipalID)) + if gen2Context.UsedPersonalMemoryFallback { + telemetry.RecordCompatibility("personal-memory-fallback", "core-runtime", "read", "gen2-context-empty", true, true, false, "USED", utterance.CorrelationID, string(session.PrincipalID), now) + } retrievedMemoryReply := interactiondomain.RecallPersonalPreference(conversationHistory, effectiveTaskMessage) stableConversationalReply, stableConversational := interactiondomain.StableConversationalReply( effectiveTaskMessage, @@ -780,9 +790,11 @@ func main() { activeFocusProvenance := situation.ProjectResolutionProvenance("") if semantic.ActiveFocus != nil { if err := repository.SaveActiveProjectFocus(r.Context(), *semantic.ActiveFocus); err != nil { + telemetry.RecordCompatibility("active-project-focus-write", "core-runtime", "write", "semantic-focus-update", true, false, true, "FAILURE", utterance.CorrelationID, string(session.PrincipalID), now) http.Error(w, "save project focus", http.StatusInternalServerError) return } + telemetry.RecordCompatibility("active-project-focus-write", "core-runtime", "write", "semantic-focus-update", true, false, true, "SUCCESS", utterance.CorrelationID, string(session.PrincipalID), now) activeFocusName = semantic.ActiveFocus.ProjectDisplayName activeFocusProvenance = semantic.ActiveFocus.Provenance } else if len(semantic.Projects) > 0 && semantic.Projects[0].Provenance == situation.ProjectResolutionInheritedFocus && previousFocus != nil { diff --git a/dev/ACTIVE_TASK.json b/dev/ACTIVE_TASK.json index 882fc61..b9e68cc 100644 --- a/dev/ACTIVE_TASK.json +++ b/dev/ACTIVE_TASK.json @@ -1 +1 @@ -{"task_file":"dev/tasks/GEN2-030-final-gen2-acceptance-documentation-baseline-and-next-evolution-queue.json","task_id":"GEN2-030","task_revision":2} +{"task_file":"dev/tasks/GEN2-COMPAT-OBS-001-production-compatibility-observation.json","task_id":"GEN2-COMPAT-OBS-001","task_revision":1} diff --git a/dev/tasks/GEN2-COMPAT-OBS-001-production-compatibility-observation.json b/dev/tasks/GEN2-COMPAT-OBS-001-production-compatibility-observation.json new file mode 100644 index 0000000..452b41f --- /dev/null +++ b/dev/tasks/GEN2-COMPAT-OBS-001-production-compatibility-observation.json @@ -0,0 +1,63 @@ +{ + "task_id": "GEN2-COMPAT-OBS-001", + "task_revision": 1, + "architecture_revision": "ARCH-GEN2-000", + "base_repo_revision": "gen2-integration", + "risk_tier": "E", + "intent": "Добавить bounded redacted production telemetry для оставшихся V1 compatibility paths без изменения их поведения, owners или canonical storage.", + "requirement_ids": ["REQ-GEN2-COMPAT-OBS-001-TELEMETRY", "REQ-GEN2-COMPAT-OBS-001-OBSERVATION", "REQ-GEN2-COMPAT-OBS-001-ROLLBACK"], + "allowed_change_scope": { + "modules": [], + "paths": [ + "platform/telemetry/**", + "cmd/core-runtime/**", + "cmd/research-runtime/**", + "modules/interaction/domain/**", + "modules/capability/domain/**", + "adapters/postgres/**", + "docs/**", + "tests/**", + "dev/ACTIVE_TASK.json", + "dev/tasks/GEN2-COMPAT-OBS-001-production-compatibility-observation.json" + ] + }, + "forbidden_change_scope": { + "paths": ["security/**", "go.mod", "go.sum", "architecture/**", ".gitea/workflows/**", "deploy/**", "database/**", "adapters/models/**"] + }, + "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 runtime logs and capability usage observation mechanisms"], + "may_change": ["telemetry event shape only; no semantic/domain contract change"] + }, + "semantic_owner": "Migration/Governance; telemetry is an operational projection and never a canonical semantic owner.", + "canonical_state": ["no change", "V1 compatibility behavior and Gen2 canonical writers remain unchanged"], + "state_transitions": ["path invocation -> redacted telemetry event -> bounded aggregation/observation -> retirement classification"], + "side_effects_and_privileges": ["stdout/stderr telemetry only; no external mutation, secret access, or user content logging"], + "persistence_and_migrations": ["no schema or migration change; telemetry is emitted to existing process logs"], + "failure_retry_restart": ["telemetry failure must never change or block the compatibility path", "rollback is code revert to the pre-instrumentation commit"], + "assumptions": ["production Supervisor deployment is authorized by this stage", "observation window is bounded and redacted", "no credentials are read"], + "acceptance_scenarios": [ + "COMPAT-OBS-001-SCN-01: each listed path has a typed redacted event shape", + "COMPAT-OBS-001-SCN-02: events include invocation, component, read/write, reason, Gen2 availability, fallback/primary, outcome and safe correlation", + "COMPAT-OBS-001-SCN-03: no event contains content, secrets, tokens or raw user identifiers", + "COMPAT-OBS-001-SCN-04: FAST tests prove instrumentation is non-blocking and path behavior is unchanged", + "COMPAT-OBS-001-SCN-05: deployed production smoke produces observable compatibility telemetry", + "COMPAT-OBS-001-EXIT: production observation can classify each path ACTIVE_REQUIRED, ACTIVE_FALLBACK, SHADOW_ONLY, UNUSED or UNKNOWN without claiming retirement from absence of evidence" + ], + "required_tests": [ + "FAST: targeted go test for changed runtime/domain/adapter packages", + "FAST: targeted go vet for changed packages", + "FAST governance: go run ./cmd/devctl preflight -active dev/ACTIVE_TASK.json", + "FAST scope: go run ./cmd/devctl check-diff -active dev/ACTIVE_TASK.json", + "FAST independent: go run ./cmd/devctl/independent", + "FAST whitespace: git diff --check gen2-integration --", + "production smoke: Supervisor candidate deployment and redacted log observation" + ], + "terminal_gap_states": ["BLOCKED_BY_SPEC_GAP", "BLOCKED_BY_ARCHITECTURE_GAP", "BLOCKED_BY_MISSING_CAPABILITY", "BLOCKED_BY_CONFLICT", "REQUIRES_MIGRATION_DECISION"] +} diff --git a/platform/telemetry/compatibility.go b/platform/telemetry/compatibility.go new file mode 100644 index 0000000..1aab50a --- /dev/null +++ b/platform/telemetry/compatibility.go @@ -0,0 +1,77 @@ +// Package telemetry contains operational projections that do not own domain +// state. Compatibility events are intentionally redacted and best-effort. +package telemetry + +import ( + "crypto/sha256" + "encoding/hex" + "encoding/json" + "log" + "strconv" + "time" +) + +// CompatibilityEvent is one bounded invocation observation. Aggregators can +// derive counts and unique request/session counts from the redacted keys. +// It deliberately contains no content, secret, or user identifier. +type CompatibilityEvent struct { + Event string `json:"event"` + Path string `json:"path"` + Component string `json:"component"` + Access string `json:"access"` + Reason string `json:"reason"` + Gen2PathAvailable bool `json:"gen2_path_available"` + Fallback bool `json:"fallback"` + Primary bool `json:"primary"` + Outcome string `json:"outcome"` + InvocationCount int `json:"invocation_count"` + UniqueRequestKey string `json:"unique_request_key"` + UniqueSessionKey string `json:"unique_session_key"` + SafeCorrelationID string `json:"correlation_id"` + LastUsedAt string `json:"last_used_at"` + ObservedAtUnixMillis int64 `json:"observed_at_unix_millis"` +} + +// NewCompatibilityEvent creates a redacted event. requestRef/sessionRef are +// hashed immediately; callers must pass identifiers only, never content. +func NewCompatibilityEvent(path, component, access, reason string, gen2PathAvailable, fallback, primary bool, outcome, requestRef, sessionRef string, now time.Time) CompatibilityEvent { + if now.IsZero() { + now = time.Now().UTC() + } + requestKey := digest(requestRef) + sessionKey := digest(sessionRef) + correlationInput := requestRef + "\x00" + sessionRef + "\x00" + path + "\x00" + strconv.FormatInt(now.UnixMilli(), 10) + return CompatibilityEvent{ + Event: "compatibility_observation", + Path: path, + Component: component, + Access: access, + Reason: reason, + Gen2PathAvailable: gen2PathAvailable, + Fallback: fallback, + Primary: primary, + Outcome: outcome, + InvocationCount: 1, + UniqueRequestKey: requestKey, + UniqueSessionKey: sessionKey, + SafeCorrelationID: digest(correlationInput), + LastUsedAt: now.UTC().Format(time.RFC3339Nano), + ObservedAtUnixMillis: now.UnixMilli(), + } +} + +// RecordCompatibility emits a structured redacted event. Telemetry is +// best-effort and cannot change the compatibility path's result. +func RecordCompatibility(path, component, access, reason string, gen2PathAvailable, fallback, primary bool, outcome, requestRef, sessionRef string, now time.Time) { + event := NewCompatibilityEvent(path, component, access, reason, gen2PathAvailable, fallback, primary, outcome, requestRef, sessionRef, now) + payload, err := json.Marshal(event) + if err != nil { + return + } + log.Printf("%s", payload) +} + +func digest(value string) string { + sum := sha256.Sum256([]byte(value)) + return hex.EncodeToString(sum[:8]) +} diff --git a/platform/telemetry/compatibility_test.go b/platform/telemetry/compatibility_test.go new file mode 100644 index 0000000..f49dfb2 --- /dev/null +++ b/platform/telemetry/compatibility_test.go @@ -0,0 +1,20 @@ +package telemetry + +import ( + "strings" + "testing" + "time" +) + +func TestCompatibilityEventRedactsRequestAndSessionReferences(t *testing.T) { + event := NewCompatibilityEvent("personal-memory-fallback", "core-runtime", "read", "empty-gen2-projection", true, true, false, "SUCCESS", "request-secret-looking", "session-secret-looking", time.UnixMilli(42)) + if event.InvocationCount != 1 || event.ObservedAtUnixMillis != 42 || event.LastUsedAt == "" { + t.Fatalf("event metadata incomplete: %+v", event) + } + if strings.Contains(event.UniqueRequestKey, "request-secret-looking") || strings.Contains(event.UniqueSessionKey, "session-secret-looking") || strings.Contains(event.SafeCorrelationID, "secret-looking") { + t.Fatalf("event leaked reference: %+v", event) + } + if event.Path != "personal-memory-fallback" || !event.Gen2PathAvailable || !event.Fallback || event.Primary { + t.Fatalf("event classification incorrect: %+v", event) + } +}