feat(store): deduplicate admitted prompts by inbox identity

docs/ARCHITECTURE.md

@@ -2,6 +2,8 @@
 
 # Architecture
 
+Local `POST /prompts` accepts optional `source_inbox_id` alongside `session_id`, `content`, and `project`. A nonempty ID identifies one prompt within its session: replay returns the existing prompt ID with the same `201` and `{"id":…, "status":"saved"}` response, without another sync mutation or write notification. Distinct IDs may contain identical text. Omitting the ID continues to append a new prompt on every call. Project ownership checks still apply before replay. Sync/import propagation and replay after deletion are not yet supported.
+
 - [How It Works](#how-it-works)
 - [Session Lifecycle](#session-lifecycle)
 - [MCP Tools](#mcp-tools)

internal/server/server.go

@@ -1051,7 +1051,7 @@ func (s *Server) handleAddPrompt(w http.ResponseWriter, r *http.Request) {
 		}
 	}
 
-	id, err := s.store.AddPrompt(body)
+	id, inserted, err := s.store.AddPromptWithResult(body)
 	if err != nil {
 		switch {
 		case errors.Is(err, store.ErrPromptContentRequired):
@@ -1063,7 +1063,9 @@ func (s *Server) handleAddPrompt(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	s.notifyWrite()
+	if inserted {
+		s.notifyWrite()
+	}
 	jsonResponse(w, http.StatusCreated, map[string]any{"id": id, "status": "saved"})
 }
 

internal/server/server_test.go

@@ -2088,6 +2088,50 @@ func TestSyncStatusResolvesProjectSelectors(t *testing.T) {
 	}
 }
 
+func TestPromptInboxIdentityHTTP(t *testing.T) {
+	st := newServerTestStore(t)
+	srv := New(st, 0)
+	h := srv.Handler()
+	var writes atomic.Int32
+	srv.SetOnWrite(func() { writes.Add(1) })
+	if err := st.CreateSession("inbox-http", "engram", "/tmp"); err != nil {
+		t.Fatal(err)
+	}
+	request := func(body string) (int, string) {
+		rec := httptest.NewRecorder()
+		h.ServeHTTP(rec, httptest.NewRequest(http.MethodPost, "/prompts", strings.NewReader(body)))
+		return rec.Code, rec.Body.String()
+	}
+	payload := `{"session_id":"inbox-http","project":"engram","content":"same","source_inbox_id":"a"}`
+	code, first := request(payload)
+	if code != http.StatusCreated || writes.Load() != 1 {
+		t.Fatalf("first: %d %s writes=%d", code, first, writes.Load())
+	}
+	var created struct {
+		ID     int64  `json:"id"`
+		Status string `json:"status"`
+	}
+	if err := json.Unmarshal([]byte(first), &created); err != nil {
+		t.Fatalf("decode first response: %v", err)
+	}
+	if created.ID <= 0 || created.Status != "saved" {
+		t.Fatalf("first response: id=%d status=%q body=%s", created.ID, created.Status, first)
+	}
+	code, replay := request(payload)
+	if code != http.StatusCreated || replay != first || writes.Load() != 1 {
+		t.Fatalf("replay: %d %s writes=%d", code, replay, writes.Load())
+	}
+	code, rejected := request(`{"session_id":"inbox-http","project":"other","content":"same","source_inbox_id":"a"}`)
+	var rejection map[string]string
+	if err := json.Unmarshal([]byte(rejected), &rejection); err != nil {
+		t.Fatalf("decode wrong-project response: %v", err)
+	}
+	if code != http.StatusBadRequest || rejection["code"] != "session_project_mismatch" ||
+		rejection["error"] != "session project does not match requested project" || writes.Load() != 1 {
+		t.Fatalf("wrong project: %d %s writes=%d", code, rejected, writes.Load())
+	}
+}
+
 // ─── OnWrite Notification Tests ──────────────────────────────────────────────
 
 func TestOnWriteCalledAfterSuccessfulWrites(t *testing.T) {

internal/store/store.go

@@ -301,9 +301,10 @@ type Prompt struct {
 }
 
 type AddPromptParams struct {
-	SessionID string `json:"session_id"`
-	Content   string `json:"content"`
-	Project   string `json:"project,omitempty"`
+	SessionID     string `json:"session_id"`
+	Content       string `json:"content"`
+	Project       string `json:"project,omitempty"`
+	SourceInboxID string `json:"source_inbox_id,omitempty"`
 }
 
 // TruncationMetadata describes storage content processing after private-tag redaction.
@@ -1234,6 +1235,7 @@ func (s *Store) migrate() error {
 			CREATE TABLE IF NOT EXISTS user_prompts (
 				id         INTEGER PRIMARY KEY AUTOINCREMENT,
 				sync_id    TEXT,
+				source_inbox_id TEXT,
 				session_id TEXT    NOT NULL,
 			content    TEXT    NOT NULL,
 			project    TEXT,
@@ -1383,6 +1385,9 @@ func (s *Store) migrate() error {
 	if err := s.addColumnIfNotExists("user_prompts", "sync_id", "TEXT"); err != nil {
 		return err
 	}
+	if err := s.addColumnIfNotExists("user_prompts", "source_inbox_id", "TEXT"); err != nil {
+		return err
+	}
 	if err := s.addColumnIfNotExists("sync_delete_tombstones", "last_remote_mutation_seq", "INTEGER"); err != nil {
 		return err
 	}
@@ -1398,6 +1403,7 @@ func (s *Store) migrate() error {
 		CREATE INDEX IF NOT EXISTS idx_obs_deleted ON observations(deleted_at);
 		CREATE INDEX IF NOT EXISTS idx_obs_dedupe ON observations(normalized_hash, project, scope, type, title, created_at DESC);
 		CREATE INDEX IF NOT EXISTS idx_prompts_sync_id ON user_prompts(sync_id);
+		CREATE UNIQUE INDEX IF NOT EXISTS idx_prompts_source_inbox ON user_prompts(session_id, source_inbox_id) WHERE source_inbox_id IS NOT NULL;
 		CREATE INDEX IF NOT EXISTS idx_prompt_tombstones_project ON prompt_tombstones(project, deleted_at DESC);
 		CREATE INDEX IF NOT EXISTS idx_sync_delete_tombstones_project ON sync_delete_tombstones(project, deleted_at DESC);
 		CREATE INDEX IF NOT EXISTS idx_sync_mutations_target_seq ON sync_mutations(target_key, seq);
@@ -3767,16 +3773,23 @@ func (s *Store) markReviewed(id int64, project string) error {
 // ─── User Prompts ────────────────────────────────────────────────────────────
 
 func (s *Store) AddPrompt(p AddPromptParams) (int64, error) {
-	// Normalize project name before storing
+	id, _, err := s.AddPromptWithResult(p)
+	return id, err
+}
+
+// AddPromptWithResult reports whether a prompt was inserted rather than replayed.
+func (s *Store) AddPromptWithResult(p AddPromptParams) (int64, bool, error) {
 	p.Project, _ = NormalizeProject(p.Project)
 
 	content, _ := s.prepareStoredContent(p.Content)
 	if content == "" {
-		return 0, ErrPromptContentRequired
+		return 0, false, ErrPromptContentRequired
 	}
 
 	var promptID int64
+	inserted := false
 	err := s.withTx(func(tx *sql.Tx) error {
+		inserted = false
 		{
 			// Settle ownership first: an unowned legacy session adopts this
 			// write's project rather than rejecting the write forever.
@@ -3787,17 +3800,29 @@ func (s *Store) AddPrompt(p AddPromptParams) (int64, error) {
 			p.Project = resolved
 		}
 		syncID := newSyncID("prompt")
-		res, err := s.execHook(tx,
-			`INSERT INTO user_prompts (sync_id, session_id, content, project) VALUES (?, ?, ?, ?)`,
-			syncID, p.SessionID, content, nullableString(p.Project),
-		)
+		query := `INSERT INTO user_prompts (sync_id, session_id, content, project, source_inbox_id) VALUES (?, ?, ?, ?, ?)`
+		if p.SourceInboxID != "" {
+			query += ` ON CONFLICT(session_id, source_inbox_id) WHERE source_inbox_id IS NOT NULL DO NOTHING`
+		}
+		res, err := s.execHook(tx, query,
+			syncID, p.SessionID, content, nullableString(p.Project), nullableString(p.SourceInboxID))
 		if err != nil {
 			return err
 		}
+		if p.SourceInboxID != "" {
+			affected, err := res.RowsAffected()
+			if err != nil {
+				return err
+			}
+			if affected == 0 {
+				return tx.QueryRow(`SELECT id FROM user_prompts WHERE session_id = ? AND source_inbox_id = ?`, p.SessionID, p.SourceInboxID).Scan(&promptID)
+			}
+		}
 		promptID, err = res.LastInsertId()
 		if err != nil {
 			return err
 		}
+		inserted = true
 		var createdAt string
 		if err := tx.QueryRow(`SELECT created_at FROM user_prompts WHERE id = ?`, promptID).Scan(&createdAt); err != nil {
 			return err
@@ -3814,9 +3839,9 @@ func (s *Store) AddPrompt(p AddPromptParams) (int64, error) {
 		})
 	})
 	if err != nil {
-		return 0, err
+		return 0, false, err
 	}
-	return promptID, nil
+	return promptID, inserted, nil
 }
 
 func (s *Store) AddPromptIfMissing(p AddPromptParams) (int64, bool, error) {