fix memory sync for interrupted turns

docs/content/building-gormes/architecture_plan/progress.json

@@ -1227,9 +1227,9 @@
             {
               "name": "Interrupted-turn memory sync suppression",
               "priority": "P2",
-              "status": "planned",
+              "status": "complete",
               "contract": "Interrupted or cancelled turns cannot flush partial observations into GONCHO or external Honcho-compatible memory",
-              "contract_status": "fixture_ready",
+              "contract_status": "validated",
               "slice_size": "small",
               "execution_owner": "memory",
               "trust_class": [
@@ -1257,7 +1257,7 @@
                 "Completed turns still sync/extract normally.",
                 "Operator status can distinguish skipped interrupted sync from extractor failures."
               ],
-              "note": "TDD: mirror upstream Hermes commit 00c3d848 by proving cancelled or interrupted turns do not flush partial memory observations. Keep this as a finalization gate over the existing GONCHO/SQLite path; do not add an external Honcho provider in this slice.",
+              "note": "TDD landed: interrupted turns are marked as skipped with reason=interrupted, stay out of extractor and Honcho-compatible turn reads, and create no goncho_conclusions; completed turns still mark ready and extract normally. Fixtures cover internal/memory/interrupted_sync_test.go plus skipped-row guards in the memory session catalog and GONCHO service. Verified with go test ./internal/kernel ./internal/memory ./internal/goncho -count=1.",
               "write_scope": [
                 "internal/kernel/",
                 "internal/memory/",

internal/goncho/service_test.go

@@ -118,6 +118,48 @@ func TestService_ContextIncludesPeerCardConclusionsAndRecentMessages(t *testing.
 	}
 }
 
+func TestService_SkipsInterruptedTurnsInSearchAndContext(t *testing.T) {
+	svc, cleanup := newTestService(t)
+	defer cleanup()
+
+	ctx := context.Background()
+	now := time.Now().Unix()
+	if _, err := svc.db.ExecContext(ctx,
+		`INSERT INTO turns(session_id, role, content, ts_unix, chat_id, memory_sync_status, memory_sync_reason)
+		 VALUES
+		 ('sess-ready', 'user', 'stable mango preference', ?, 'telegram:6586915095', 'ready', NULL),
+		 ('sess-skip', 'user', 'interrupted pineapple draft', ?, 'telegram:6586915095', 'skipped', 'interrupted')`,
+		now, now+1,
+	); err != nil {
+		t.Fatal(err)
+	}
+
+	search, err := svc.Search(ctx, SearchParams{
+		Peer:       "telegram:6586915095",
+		Query:      "pineapple",
+		MaxTokens:  200,
+		SessionKey: "telegram:6586915095",
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(search.Results) != 0 {
+		t.Fatalf("Search returned skipped turn results: %+v", search.Results)
+	}
+
+	got, err := svc.Context(ctx, ContextParams{
+		Peer:       "telegram:6586915095",
+		MaxTokens:  400,
+		SessionKey: "telegram:6586915095",
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(got.RecentMessages) != 1 || got.RecentMessages[0].Content != "stable mango preference" {
+		t.Fatalf("RecentMessages = %+v, want only ready turn", got.RecentMessages)
+	}
+}
+
 func TestService_DeleteConclusion(t *testing.T) {
 	svc, cleanup := newTestService(t)
 	defer cleanup()

internal/goncho/sql.go

@@ -163,6 +163,7 @@ func findTurns(ctx context.Context, db *sql.DB, query, sessionKey string, limit
 		SELECT content
 		FROM turns
 		WHERE (chat_id = ? OR session_id = ?)
+		  AND memory_sync_status = 'ready'
 	`
 	args := []any{sessionKey, sessionKey}
 	if trimmed := strings.TrimSpace(query); trimmed != "" {
@@ -200,7 +201,8 @@ func recentTurns(ctx context.Context, db *sql.DB, sessionKey string, limit int)
 	rows, err := db.QueryContext(ctx, `
 		SELECT role, content
 		FROM turns
-		WHERE chat_id = ? OR session_id = ?
+		WHERE (chat_id = ? OR session_id = ?)
+		  AND memory_sync_status = 'ready'
 		ORDER BY ts_unix DESC, id DESC
 		LIMIT ?
 	`, sessionKey, sessionKey, limit)

internal/kernel/kernel.go

@@ -67,6 +67,10 @@ type SkillUsageRecorder interface {
 	RecordSkillUsage(ctx context.Context, skillNames []string) error
 }
 
+type memorySyncSkipper interface {
+	SkipMemorySync(ctx context.Context, turnKey, reason string) error
+}
+
 type Kernel struct {
 	cfg    Config
 	client hermes.Client
@@ -211,6 +215,7 @@ func (k *Kernel) Run(ctx context.Context) error {
 // to the store.Command payload and is otherwise opaque to the kernel.
 func (k *Kernel) runTurn(ctx context.Context, text, sessionContext, cronJobID string) {
 	prov := newProvenance(k.cfg.Endpoint)
+	turnKey := prov.LocalRunID
 
 	// 1. Admission. Reject locally before any HTTP.
 	if err := k.cfg.Admission.Validate(text); err != nil {
@@ -223,12 +228,14 @@ func (k *Kernel) runTurn(ctx context.Context, text, sessionContext, cronJobID st
 	// 2. Persist user turn with hard 250ms ack deadline (spec §7.8 store row).
 	storeCtx, storeCancel := context.WithTimeout(ctx, StoreAckDeadline)
 	userPayload, _ := json.Marshal(map[string]any{
-		"session_id":  k.sessionID,
-		"content":     text,
-		"ts_unix":     time.Now().Unix(),
-		"chat_id":     k.cfg.ChatKey,
-		"cron":        cronFlag(cronJobID),
-		"cron_job_id": cronJobID,
+		"session_id":         k.sessionID,
+		"content":            text,
+		"ts_unix":            time.Now().Unix(),
+		"chat_id":            k.cfg.ChatKey,
+		"cron":               cronFlag(cronJobID),
+		"cron_job_id":        cronJobID,
+		"turn_key":           turnKey,
+		"memory_sync_status": "pending",
 	})
 	_, err := k.store.Exec(storeCtx, store.Command{Kind: store.AppendUserTurn, Payload: userPayload})
 	storeCancel()
@@ -483,6 +490,7 @@ toolLoop:
 	}
 
 	if cancelled {
+		k.skipMemorySync(turnKey, "interrupted")
 		k.phase = PhaseCancelling
 		k.emitFrame("cancelled")
 	} else if k.draft != "" {
@@ -491,10 +499,12 @@ toolLoop:
 		// handles I/O off the hot path. 250ms context bound kept as a safety net
 		// in case someone injects a synchronous store in the future.
 		payload := map[string]any{
-			"session_id": k.sessionID,
-			"content":    k.draft,
-			"ts_unix":    time.Now().Unix(),
-			"chat_id":    k.cfg.ChatKey,
+			"session_id":         k.sessionID,
+			"content":            k.draft,
+			"ts_unix":            time.Now().Unix(),
+			"chat_id":            k.cfg.ChatKey,
+			"turn_key":           turnKey,
+			"memory_sync_status": "ready",
 		}
 		if len(toolCallsSeen) > 0 {
 			meta, _ := json.Marshal(map[string]any{"tool_calls": toolCallsSeen})
@@ -724,6 +734,18 @@ func cronFlag(cronJobID string) int {
 	return 1
 }
 
+func (k *Kernel) skipMemorySync(turnKey, reason string) {
+	skipper, ok := k.store.(memorySyncSkipper)
+	if !ok || turnKey == "" {
+		return
+	}
+	skipCtx, cancel := context.WithTimeout(context.Background(), StoreAckDeadline)
+	defer cancel()
+	if err := skipper.SkipMemorySync(skipCtx, turnKey, reason); err != nil {
+		k.log.Warn("kernel: skip memory sync failed", "err", err)
+	}
+}
+
 // truncate returns s clamped to n runes with an ellipsis suffix. Safe on
 // non-ASCII input.
 func truncate(s string, n int) string {

internal/memory/extractor.go

@@ -181,7 +181,7 @@ func (e *Extractor) loopOnce(ctx context.Context) {
 func (e *Extractor) pollBatch(ctx context.Context) ([]turnRow, error) {
 	rows, err := e.store.db.QueryContext(ctx,
 		`SELECT id, role, content FROM turns
-		 WHERE extracted = 0 AND cron = 0 AND extraction_attempts < ?
+		 WHERE extracted = 0 AND cron = 0 AND memory_sync_status = 'ready' AND extraction_attempts < ?
 		 ORDER BY id LIMIT ?`,
 		e.cfg.MaxAttempts, e.cfg.BatchSize)
 	if err != nil {