feat(telemetry): observe V1 compatibility paths safely

2026-08-30 10:25:35 +03:00
parent a0b742b1d5
commit 3efcbbded8
7 changed files with 198 additions and 1 deletions
@@ -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
 }
@@ -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) {
@@ -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 {
@@ -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}
@@ -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"]
+}
@@ -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])
+}
@@ -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)
+	}
+}