From a6be96410cff43359699ba89d62ccef3bf309b5e Mon Sep 17 00:00:00 2001 From: Shay Date: Fri, 15 May 2026 21:51:16 -0700 Subject: [PATCH] fix(session): CGA-only hemisphere consistency, remove forbidden Euclidean metric MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit _orient_result_to_anchor used np.dot (Euclidean dot product) alongside cga_inner to decide hemisphere flips. When CGA inner was positive (correct hemisphere) but Euclidean was negative, the flip negated CGA alignment — making correctly-oriented fields rank last in vault recall. Changes: - Move hemisphere check into finalize_turn so all paths (ChatRuntime, SessionContext.respond) get consistent protection. - Use CGA inner product only, removing the forbidden Euclidean metric. - Remove _orient_result_to_anchor (subsumed by finalize_turn). - Remove SessionContext.arespond (dead code, no callers). Co-Authored-By: Claude Opus 4.6 --- session/context.py | 62 +++++++++++---------------------- tests/test_session_coherence.py | 29 +++++++++++++++ 2 files changed, 49 insertions(+), 42 deletions(-) diff --git a/session/context.py b/session/context.py index de439c56..121ff344 100644 --- a/session/context.py +++ b/session/context.py @@ -145,6 +145,21 @@ class SessionContext: pass self.referents.register_from_tokens(result.tokens, versors, turn=self.turn) + def _hemisphere_consistent_field(self, field_state: FieldState) -> FieldState: + """Ensure field stays in the same CGA hemisphere as the session anchor.""" + if self._anchor_field is None: + return field_state + if cga_inner(field_state.F, self._anchor_field) >= 0.0: + return field_state + return FieldState( + F=-field_state.F, + node=field_state.node, + step=field_state.step, + holonomy=field_state.holonomy, + energy=field_state.energy, + valence=field_state.valence, + ) + def finalize_turn( self, result: GenerationResult, @@ -168,24 +183,25 @@ class SessionContext: active_slots = self.referents.active_slots() self._register_result_referent(result) - # Include any newly registered output referent in the turn metadata. active_slots = self.referents.active_slots() | active_slots + oriented_state = self._hemisphere_consistent_field(result.final_state) + self.graph.add_turn( turn_idx=self.turn, input_versor=input_F, - output_versor=result.final_state.F, + output_versor=oriented_state.F, tokens_in=turn_tokens, tokens_out=tuple(result.tokens or []), dialogue_role=dialogue_role, referent_slots=active_slots, backward_edges=backward_edges, ) - self.state = result.final_state + self.state = oriented_state payload = {"turn": self.turn, "role": "assistant"} if metadata: payload.update(metadata) - self.vault.store(result.final_state.F, payload) + self.vault.store(oriented_state.F, payload) self.turn += 1 self._last_response_tokens = result.tokens @@ -228,47 +244,9 @@ class SessionContext: valence=self.state.valence, ) result = generate(pivot, self.vocab, self.persona, max_tokens, vault=self.vault) - result = self._orient_result_to_anchor(result) self.finalize_turn(result, input_versor=input_versor, dialogue_role="assert") return result - def _orient_result_to_anchor(self, result: GenerationResult) -> GenerationResult: - final_state = result.final_state - coherence_anchor = self._anchor_field if self._anchor_field is not None else (self.state.F if self.state is not None else None) - if coherence_anchor is None: - return result - cga_score = cga_inner(final_state.F, coherence_anchor) - euclidean_score = float(np.dot(final_state.F, coherence_anchor)) - if cga_score < 0.0 or euclidean_score < 0.0: - final_state = FieldState( - F=-final_state.F, - node=final_state.node, - step=final_state.step, - holonomy=final_state.holonomy, - energy=final_state.energy, - valence=final_state.valence, - ) - return GenerationResult( - tokens=result.tokens, - final_state=final_state, - trajectory=result.trajectory, - salience_top_k=result.salience_top_k, - candidates_used=result.candidates_used, - vault_hits=result.vault_hits, - identity_score=result.identity_score, - ) - return result - - async def arespond(self, max_tokens: int = 128): - assert self.state is not None, "Call ingest() before arespond()." - input_versor = self._last_input_versor.copy() if self._last_input_versor is not None else self.state.F.copy() - result = self._orient_result_to_anchor( - generate(self.state, self.vocab, self.persona, max_tokens, vault=self.vault) - ) - for token in result.tokens: - yield token - self.finalize_turn(result, input_versor=input_versor, dialogue_role="assert") - def recall(self, query_tokens: list, top_k: int = 5) -> list: query_state = inject(query_tokens, self.vocab) return self.vault.recall(query_state.F, top_k=top_k) diff --git a/tests/test_session_coherence.py b/tests/test_session_coherence.py index 960058c6..a4d8b6fe 100644 --- a/tests/test_session_coherence.py +++ b/tests/test_session_coherence.py @@ -57,6 +57,35 @@ def _farther_unrelated(result_F: np.ndarray, prompt_F: np.ndarray, start_seed: i raise AssertionError("Could not construct a deterministic farther unrelated versor.") +def test_finalize_turn_enforces_cga_hemisphere_consistency() -> None: + """Vault entries must share the anchor's CGA hemisphere for recall to rank correctly.""" + from generate.result import GenerationResult + from field.state import FieldState + + vocab = _vocab() + session = SessionContext(vocab=vocab) + session.ingest(["logos"]) + anchor = session._anchor_field.copy() + + anti_F = -session.state.F + anti_state = FieldState( + F=anti_F, + node=session.state.node, + step=session.state.step, + holonomy=session.state.holonomy, + energy=session.state.energy, + valence=session.state.valence, + ) + anti_result = GenerationResult(tokens=("arche",), final_state=anti_state, vault_hits=0) + assert cga_inner(anti_F, anchor) < 0.0, "precondition: anti-hemisphere field" + + session.finalize_turn(anti_result, dialogue_role="assert") + + assert cga_inner(session.state.F, anchor) >= 0.0, ( + "finalize_turn must flip anti-hemisphere fields to keep vault recall consistent" + ) + + def test_repeated_prompt_accumulates_field_and_stays_prompt_coherent() -> None: session = SessionContext(vocab=_vocab()) prompt = ["logos", "arche"]