Fix memory extraction misattribution, double-extraction, and contradiction bugs

- Skip speaker_info label for pre-labeled history messages (starting with @)
  in _format_conversation to prevent double-labeling misattribution
- Add touch_debounce() to MemoryPostprocessService and wire it through RPC
  extraction handlers to prevent postprocess double-extraction
- Add _conflicts_with_self_fact() guard in process_extracted_facts() to drop
  third-party person_facts that contradict authoritative self-facts

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: dcramer

src/ash/integrations/memory.py

@@ -52,6 +52,7 @@ def register_rpc_methods(self, server, context: IntegrationContext) -> None:
             components.memory_manager,
             memory_extractor=components.memory_extractor,
             sessions_path=context.sessions_path,
+            postprocess_service=self._postprocess,
         )
 
     async def on_message_postprocess(

src/ash/memory/extractor.py

@@ -756,7 +756,9 @@ def _format_conversation(
                 text = text[:2000] + "..."
 
             if msg.role == Role.USER:
-                if speaker_info:
+                # Skip speaker_info for pre-labeled history messages (already
+                # contain @username: prefix from chat history loading).
+                if speaker_info and not text.lstrip().startswith("@"):
                     label = speaker_info.format_label()
                     lines.append(f"<user>\n{label}: {text}\n</user>")
                 else:

src/ash/memory/postprocess.py

@@ -48,6 +48,14 @@ def __init__(
         self._confidence_threshold = confidence_threshold
         self._last_extraction_time: float | None = None
 
+    def touch_debounce(self) -> None:
+        """Mark an extraction as having just occurred.
+
+        Called by RPC extraction handlers so the postprocess debounce timer
+        is aware that extraction already happened, preventing double-extraction.
+        """
+        self._last_extraction_time = time.time()
+
     def maybe_schedule(
         self,
         *,

src/ash/memory/processing.py

@@ -351,6 +351,75 @@ async def build_owner_names_for_speaker(
     return owner_names
 
 
+async def _conflicts_with_self_fact(
+    store: Store,
+    content: str,
+    subject_person_ids: list[str],
+    speaker_person_id: str | None,
+) -> bool:
+    """Check if a person_fact conflicts with an existing authoritative self-fact.
+
+    When a third party claims something about a subject (person_fact), and the
+    subject already has a self-fact (stated by themselves) on the same topic,
+    the third-party claim should be dropped to preserve subject authority.
+
+    Only applies when the speaker is NOT the subject (i.e. third-party claims).
+    """
+    from ash.graph.edges import get_memories_about_person
+    from ash.store.trust import classify_trust
+    from ash.store.types import get_assertion
+
+    if not subject_person_ids:
+        return False
+
+    # If the speaker is the subject, this is a self-fact update, not a conflict
+    if speaker_person_id and speaker_person_id in subject_person_ids:
+        return False
+
+    # Find existing self-facts about each subject
+    for pid in subject_person_ids:
+        memory_ids = get_memories_about_person(store._graph, pid)
+        for mid in memory_ids:
+            memory = store._graph.memories.get(mid)
+            if not memory or memory.superseded_at or memory.archived_at:
+                continue
+
+            # Must be a self-fact (speaker is the subject)
+            trust = classify_trust(store._graph, mid)
+            if trust != "fact":
+                continue
+
+            # Also verify assertion kind is SELF_FACT if assertion exists
+            assertion = get_assertion(memory)
+            if assertion and assertion.assertion_kind != AssertionKind.SELF_FACT:
+                continue
+
+            # Check semantic similarity — is the new claim about the same topic?
+            try:
+                query_embedding = await store._embeddings.embed(content)
+                similar = store._index.search(query_embedding, limit=5)
+                for found_id, similarity in similar:
+                    if found_id == mid and similarity >= 0.75:
+                        logger.info(
+                            "person_fact_blocked_by_self_fact",
+                            extra={
+                                "fact.content": content[:80],
+                                "self_fact.id": mid,
+                                "self_fact.content": memory.content[:80],
+                                "similarity": similarity,
+                            },
+                        )
+                        return True
+            except Exception:
+                logger.debug(
+                    "self_fact_conflict_check_failed",
+                    extra={"person_id": pid},
+                    exc_info=True,
+                )
+
+    return False
+
+
 async def process_extracted_facts(
     facts: list[ExtractedFact],
     store: Store,
@@ -536,6 +605,17 @@ async def process_extracted_facts(
                 )
                 assertion = downgrade_assertion_to_context(assertion)
 
+            # Guard: drop third-party person_facts that contradict authoritative
+            # self-facts from the subject. See specs/memory/index.md.
+            if (
+                assertion.assertion_kind == AssertionKind.PERSON_FACT
+                and subject_person_ids
+                and await _conflicts_with_self_fact(
+                    store, fact.content, subject_person_ids, stated_by_pid
+                )
+            ):
+                continue
+
             # DM sensitivity floor: ephemeral types get minimum PERSONAL
             # in private chats as defense-in-depth against cross-context leakage
             effective_sensitivity = fact.sensitivity

src/ash/rpc/methods/memory.py

@@ -18,6 +18,7 @@
 
 if TYPE_CHECKING:
     from ash.memory.extractor import MemoryExtractor
+    from ash.memory.postprocess import MemoryPostprocessService
     from ash.rpc.server import RPCServer
     from ash.store.store import Store
 
@@ -31,6 +32,7 @@ def register_memory_methods(
     memory_manager: "Store",
     memory_extractor: "MemoryExtractor | None" = None,
     sessions_path: Path | None = None,
+    postprocess_service: "MemoryPostprocessService | None" = None,
 ) -> None:
     """Register memory-related RPC methods.
 
@@ -39,6 +41,7 @@ def register_memory_methods(
         memory_manager: Store instance.
         memory_extractor: Optional extractor for fact classification/extraction.
         sessions_path: Path to sessions directory (for memory.extract).
+        postprocess_service: Optional postprocess service for debounce coordination.
     """
 
     async def _build_username_lookup() -> dict[str, str]:
@@ -496,6 +499,11 @@ async def _extract_and_store_from_messages(
             chat_type=chat_type,
         )
 
+        # Touch postprocess debounce so the background extraction timer
+        # knows an RPC extraction just occurred, preventing double-extraction.
+        if postprocess_service and stored_ids:
+            postprocess_service.touch_debounce()
+
         return {"stored": len(stored_ids)}
 
     async def memory_extract_from_messages(params: dict[str, Any]) -> dict[str, Any]:

tests/test_memory_extractor.py

@@ -961,3 +961,51 @@ def test_aliases_skips_invalid_inner_types(self, extractor):
 
         assert len(facts) == 1
         assert facts[0].aliases == {}
+
+
+class TestFormatConversation:
+    """Tests for _format_conversation speaker labeling behavior."""
+
+    @pytest.fixture
+    def extractor(self):
+        return MemoryExtractor(
+            llm=MagicMock(),
+            model="test-model",
+            confidence_threshold=0.7,
+        )
+
+    def test_skips_speaker_info_for_at_prefixed_messages(self, extractor):
+        """Pre-labeled history messages (starting with @) should not get speaker_info prepended."""
+        speaker = SpeakerInfo(username="sksembhi", display_name="SK")
+        messages = [
+            Message(role=Role.USER, content="@evanpurkhiser (Evan): I'm 6'2\""),
+        ]
+        result = extractor._format_conversation(messages, speaker_info=speaker)
+        # Should NOT prepend @sksembhi label — message is already labeled
+        assert "@sksembhi" not in result
+        assert "@evanpurkhiser (Evan): I'm 6'2\"" in result
+
+    def test_adds_speaker_info_for_unprefixed_messages(self, extractor):
+        """Unprefixed user messages should get speaker_info label prepended."""
+        speaker = SpeakerInfo(username="sksembhi", display_name="SK")
+        messages = [
+            Message(role=Role.USER, content="Hello world"),
+        ]
+        result = extractor._format_conversation(messages, speaker_info=speaker)
+        assert "@sksembhi (SK): Hello world" in result
+
+    def test_mixed_history_and_current_messages(self, extractor):
+        """History (pre-labeled) and current (unlabeled) messages should be handled correctly."""
+        speaker = SpeakerInfo(username="sksembhi", display_name="SK")
+        messages = [
+            Message(role=Role.USER, content="@evanpurkhiser (Evan): I'm 6'2\""),
+            Message(role=Role.ASSISTANT, content="Got it!"),
+            Message(role=Role.USER, content="Evan's height is 5'2\""),
+        ]
+        result = extractor._format_conversation(messages, speaker_info=speaker)
+        # History message: no speaker_info prepended
+        assert "@sksembhi" not in result.split("</user>")[0]
+        # Current message: speaker_info prepended
+        assert "@sksembhi (SK): Evan's height is 5'2\"" in result
+        # History message preserved
+        assert "@evanpurkhiser (Evan): I'm 6'2\"" in result