From b1870508292c69ab5057bed9ec7f4f6e2f27827f Mon Sep 17 00:00:00 2001 From: Jiri Puc Date: Tue, 28 Jul 2026 23:38:07 +0200 Subject: [PATCH 1/2] feat(e0): implement D79 summaries and placement --- .env.example | 1 + compose.yaml | 1 + plan/designs/e0_files_design.md | 9 +- src/rememberstack/model/__init__.py | 4 + src/rememberstack/model/sections.py | 19 +- src/rememberstack/profiles/selfhost.py | 5 + src/rememberstack/spine/document_catalog.py | 154 +++- src/rememberstack/workers/__init__.py | 6 + src/rememberstack/workers/e0.py | 289 ++++-- src/rememberstack/workers/e0_summary.py | 823 ++++++++++++++++++ src/tests/profiles/test_selfhost_profile.py | 14 + src/tests/spine/test_forget_catalog.py | 54 +- src/tests/workers/test_d79_structure_route.py | 41 +- src/tests/workers/test_d79_summaries.py | 220 +++++ src/tests/workers/test_e0_chain.py | 340 +++++++- website/src/app/docs/deployment/page.mdx | 12 +- 16 files changed, 1825 insertions(+), 167 deletions(-) create mode 100644 src/rememberstack/workers/e0_summary.py create mode 100644 src/tests/workers/test_d79_summaries.py diff --git a/.env.example b/.env.example index f6d719ac..4b5c96de 100644 --- a/.env.example +++ b/.env.example @@ -42,3 +42,4 @@ REMEMBERSTACK_OPENROUTER_API_KEY=replace-before-real-use # REMEMBERSTACK_STRUCTURER_MODEL=openai/gpt-5.6-luna # REMEMBERSTACK_SKELETON_CHECK_MODEL=z-ai/glm-4.7-flash # REMEMBERSTACK_ROLE_MODEL=z-ai/glm-4.7-flash +# REMEMBERSTACK_SUMMARY_MODEL=z-ai/glm-4.7-flash diff --git a/compose.yaml b/compose.yaml index dd4826b1..037a6f32 100644 --- a/compose.yaml +++ b/compose.yaml @@ -27,6 +27,7 @@ x-app: &app REMEMBERSTACK_STRUCTURER_MODEL: ${REMEMBERSTACK_STRUCTURER_MODEL:-openai/gpt-5.6-luna} REMEMBERSTACK_SKELETON_CHECK_MODEL: ${REMEMBERSTACK_SKELETON_CHECK_MODEL:-z-ai/glm-4.7-flash} REMEMBERSTACK_ROLE_MODEL: ${REMEMBERSTACK_ROLE_MODEL:-z-ai/glm-4.7-flash} + REMEMBERSTACK_SUMMARY_MODEL: ${REMEMBERSTACK_SUMMARY_MODEL:-z-ai/glm-4.7-flash} REMEMBERSTACK_E1_EMBEDDING_MODEL: ${REMEMBERSTACK_E1_EMBEDDING_MODEL:-qwen/qwen3-embedding-8b} REMEMBERSTACK_E1_PREFIX_MODEL: ${REMEMBERSTACK_E1_PREFIX_MODEL:-openai/gpt-5.6-luna} REMEMBERSTACK_E2_EXTRACT_MODEL: ${REMEMBERSTACK_E2_EXTRACT_MODEL:-openai/gpt-5.6-luna} diff --git a/plan/designs/e0_files_design.md b/plan/designs/e0_files_design.md index b1827342..dcd71ab3 100644 --- a/plan/designs/e0_files_design.md +++ b/plan/designs/e0_files_design.md @@ -385,7 +385,14 @@ unchanged): parallel and cached per section; the cache key is the ordered constituent block hashes, child-summary hashes, model, prompt, generation parameters, and summarizer version. This is the call shape where cheap models are reliable: bounded context, no arithmetic, no giant - output. + output. **Degradation convention:** `summary_version` is non-null only when every section + in the generation has a summary; any failed section call leaves that section null (and any + ancestor that cannot receive all child one-liners null), may preserve successful summaries + in independent subtrees, and makes the generation's `summary_version` null. Because root + placement is emitted by the same complete reduction, `placement_version` is non-null iff + `summary_version` and root `placement_path` are non-null; otherwise both placement fields + are null. Thus a partially useful tree is persisted without mislabeling a degraded run as a + complete summary generation. - **Placement rides the root reduction.** The one-shot call also produced the D39 placement hint; that responsibility moves to the **document-level (root) summary call**, which sees exactly what placement needs — the title, source kind, and the child one-liners. On the diff --git a/src/rememberstack/model/__init__.py b/src/rememberstack/model/__init__.py index 346e93f0..28d074c0 100644 --- a/src/rememberstack/model/__init__.py +++ b/src/rememberstack/model/__init__.py @@ -302,6 +302,8 @@ from rememberstack.model.sections import ProposedSection from rememberstack.model.sections import RoleAssignment from rememberstack.model.sections import RoleClassificationResponse +from rememberstack.model.sections import RootSummaryPlacementResponse +from rememberstack.model.sections import SectionSummaryResponse from rememberstack.model.sections import SectionTreeRecord from rememberstack.model.sections import SkeletonCheckOutcome from rememberstack.model.sections import SkeletonCheckRecord @@ -591,7 +593,9 @@ "ProposedSection", "RoleAssignment", "RoleClassificationResponse", + "RootSummaryPlacementResponse", "SectionTreeRecord", + "SectionSummaryResponse", "SkeletonCheckOutcome", "SkeletonCheckRecord", "SkeletonCheckResponse", diff --git a/src/rememberstack/model/sections.py b/src/rememberstack/model/sections.py index 84975212..c78a89ee 100644 --- a/src/rememberstack/model/sections.py +++ b/src/rememberstack/model/sections.py @@ -164,6 +164,23 @@ class RoleClassificationResponse(BaseModel): assignments: tuple[RoleAssignment, ...] = () +class SectionSummaryResponse(BaseModel): + """One bounded, single-line section summary and no auxiliary prose.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + summary: str = Field(min_length=1, max_length=512, pattern=r"^[^\r\n]+$") + + +class RootSummaryPlacementResponse(BaseModel): + """The root reduction's closed two-field summary + D39 placement shape.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + summary: str = Field(min_length=1, max_length=512, pattern=r"^[^\r\n]+$") + placement_path: str = Field(min_length=3, max_length=512, pattern=r"^/[^\r\n]+/$") + + class SnappedSection(BaseModel): """One well-formed section after the deterministic snap (block coordinates). @@ -181,7 +198,7 @@ class SnappedSection(BaseModel): block_end: int = Field(ge=-1) char_start: int = Field(ge=0) char_end: int = Field(ge=0) - summary: str + summary: str | None ordinal: int = Field(ge=0) heading_level: int | None = Field(default=None, ge=1, le=6) normalized_title: str = "" diff --git a/src/rememberstack/profiles/selfhost.py b/src/rememberstack/profiles/selfhost.py index ad42538a..9a2ef9b5 100644 --- a/src/rememberstack/profiles/selfhost.py +++ b/src/rememberstack/profiles/selfhost.py @@ -391,6 +391,7 @@ def _handler(self, *, stage: PipelineStage) -> StageHandler: from rememberstack.workers import SkeletonCheckSettings from rememberstack.workers import StructureHandler from rememberstack.workers import StructurerSettings + from rememberstack.workers import SummarySettings documents = DocumentCatalog(engine=self._engine) chunks = ChunkCatalog(engine=self._engine) @@ -417,6 +418,7 @@ def _handler(self, *, stage: PipelineStage) -> StageHandler: settings=StructurerSettings.model_validate({}), check_settings=SkeletonCheckSettings.model_validate({}), role_settings=RoleSettings.model_validate({}), + summary_settings=SummarySettings.model_validate({}), ) if stage is PipelineStage.CHUNK: return ChunkHandler( @@ -598,10 +600,12 @@ def _model_bindings() -> dict[str, str]: from rememberstack.workers import RoleSettings from rememberstack.workers import SkeletonCheckSettings from rememberstack.workers import StructurerSettings + from rememberstack.workers import SummarySettings structurer = StructurerSettings.model_validate({}) skeleton_check = SkeletonCheckSettings.model_validate({}) roles = RoleSettings.model_validate({}) + summaries = SummarySettings.model_validate({}) e1 = E1Settings.model_validate({}) e2 = E2Settings.model_validate({}) e3 = E3Settings.model_validate({}) @@ -613,6 +617,7 @@ def _model_bindings() -> dict[str, str]: "structure_fallback": structurer.model, "skeleton_check": skeleton_check.model, "section_role": roles.model, + "section_summary": summaries.model, "chunk_embedding": e1.embedding_model, "context_prefix": e1.prefix_model, "claim_extraction": e2.extract_model, diff --git a/src/rememberstack/spine/document_catalog.py b/src/rememberstack/spine/document_catalog.py index 92e0cf6c..365f635a 100644 --- a/src/rememberstack/spine/document_catalog.py +++ b/src/rememberstack/spine/document_catalog.py @@ -7,6 +7,7 @@ skeleton checks append independently under D52. """ +from collections.abc import Sequence from decimal import Decimal import json from uuid import NAMESPACE_URL @@ -17,6 +18,7 @@ from sqlalchemy import text from sqlalchemy.engine import Connection from sqlalchemy.engine import Engine +from sqlalchemy.engine import RowMapping from rememberstack.model import ConvertSource from rememberstack.model import DocumentVersionNotFoundError @@ -215,6 +217,44 @@ def structure_source(self, *, representation_id: UUID) -> StructureSource: ) return StructureSource.model_validate(dict(row)) + def current_section_tree( + self, *, representation_id: UUID + ) -> PersistedSectionTree | None: + """Load the representation's current immutable structure generation.""" + with self._engine.connect() as connection: + generation = ( + connection.execute( + _SELECT_CURRENT_STRUCTURE_GENERATION, + {"representation_id": representation_id}, + ) + .mappings() + .one_or_none() + ) + if generation is None: + return None + persisted = ( + connection.execute( + _SELECT_SECTION_TREE, + {"structure_generation_id": generation["structure_generation_id"]}, + ) + .mappings() + .all() + ) + if not persisted: + raise RuntimeError("current structure generation has no root section") + return _persisted_tree(generation=generation, sections=persisted) + + def summary_cache_sidecars( + self, *, doc_id: UUID, summary_version: str + ) -> tuple[str, ...]: + """Prior successful summary generations for sidecar cache lookup.""" + with self._engine.connect() as connection: + rows = connection.execute( + _SELECT_SUMMARY_CACHE_SIDECARS, + {"doc_id": doc_id, "summary_version": summary_version}, + ).scalars() + return tuple(str(uri) for uri in rows) + def record_synthetic_root(self, *, record: SyntheticRootRecord) -> None: """Compatibility helper for callers that explicitly request a root. @@ -446,45 +486,7 @@ def record_section_tree(self, *, record: SectionTreeRecord) -> PersistedSectionT _SUPERSEDE_PRIOR_VERSIONS, # are superseded as of now (D55) {"doc_id": record.doc_id, "version_id": record.version_id}, ) - return PersistedSectionTree( - sections=tuple( - SnappedSection( - node_path=row["node_path"], - parent_path=( - row["node_path"].rsplit(".", 1)[0] - if "." in row["node_path"] - else None - ), - title=row["title"] or "", - role=row["role"], - block_start=row["block_start"], - block_end=row["block_end"], - char_start=row["char_start"], - char_end=row["char_end"], - summary=row["summary"] or "", - ordinal=row["ordinal"], - heading_level=row["heading_level"], - normalized_title=row["normalized_title"], - ) - for row in persisted - ), - placement_path=persisted[0]["placement_path"], - structurer_version=persisted[0]["structurer_version"] or "", - structure_generation_id=generation["structure_generation_id"], - pageindex_uri=generation["pageindex_uri"], - skeleton_version=generation["skeleton_version"], - skeleton_hash=generation["skeleton_hash"], - skeleton_producer_family=generation["skeleton_producer_family"], - skeleton_check_version=generation["skeleton_check_version"], - roles_version=generation["roles_version"], - summary_version=generation["summary_version"], - placement_version=generation["placement_version"], - selecting_check_id=generation["selecting_check_id"], - route_tag=generation["route_tag"], - candidate_skeleton_hash=generation["candidate_skeleton_hash"], - stats_version=generation["stats_version"], - stats=SkeletonStats.model_validate(generation["stats"]), - ) + return _persisted_tree(generation=generation, sections=persisted) def _lineage_locked(*, connection: Connection, record: UploadRecord) -> UUID: @@ -732,6 +734,32 @@ def _lineage_locked(*, connection: Connection, record: UploadRecord) -> UUID: """ ) +_SELECT_CURRENT_STRUCTURE_GENERATION = text( + """ + SELECT g.structure_generation_id, g.skeleton_version, g.skeleton_hash, + g.skeleton_producer_family, g.skeleton_check_version, g.roles_version, + g.summary_version, g.placement_version, g.selecting_check_id, + g.route_tag::text AS route_tag, g.candidate_skeleton_hash, + g.stats_version, g.stats, g.pageindex_uri + FROM document_representations r + JOIN document_structure_generations g + ON g.representation_id = r.representation_id + AND g.structure_generation_id = r.current_structure_generation_id + WHERE r.representation_id = :representation_id + """ +) + +_SELECT_SUMMARY_CACHE_SIDECARS = text( + """ + SELECT pageindex_uri + FROM document_structure_generations + WHERE doc_id = :doc_id + AND summary_version = :summary_version + AND pageindex_uri IS NOT NULL + ORDER BY created_at DESC, structure_generation_id + """ +) + _SELECT_SECTION_TREE = text( """ SELECT node_path, title, role::text AS role, block_start, block_end, @@ -743,6 +771,54 @@ def _lineage_locked(*, connection: Connection, record: UploadRecord) -> UUID: """ ) + +def _persisted_tree( + *, generation: RowMapping, sections: Sequence[RowMapping] +) -> PersistedSectionTree: + """Materialize one immutable generation from SQL mapping rows.""" + generation_row = generation + section_rows = sections + return PersistedSectionTree( + sections=tuple( + SnappedSection( + node_path=row["node_path"], + parent_path=( + row["node_path"].rsplit(".", 1)[0] + if "." in row["node_path"] + else None + ), + title=row["title"] or "", + role=row["role"], + block_start=row["block_start"], + block_end=row["block_end"], + char_start=row["char_start"], + char_end=row["char_end"], + summary=row["summary"], + ordinal=row["ordinal"], + heading_level=row["heading_level"], + normalized_title=row["normalized_title"], + ) + for row in section_rows + ), + placement_path=section_rows[0]["placement_path"], + structurer_version=section_rows[0]["structurer_version"] or "", + structure_generation_id=generation_row["structure_generation_id"], + pageindex_uri=generation_row["pageindex_uri"] or "", + skeleton_version=generation_row["skeleton_version"], + skeleton_hash=generation_row["skeleton_hash"], + skeleton_producer_family=generation_row["skeleton_producer_family"], + skeleton_check_version=generation_row["skeleton_check_version"], + roles_version=generation_row["roles_version"], + summary_version=generation_row["summary_version"], + placement_version=generation_row["placement_version"], + selecting_check_id=generation_row["selecting_check_id"], + route_tag=generation_row["route_tag"], + candidate_skeleton_hash=generation_row["candidate_skeleton_hash"], + stats_version=generation_row["stats_version"], + stats=SkeletonStats.model_validate(generation_row["stats"]), + ) + + _MARK_REPRESENTATION_READY = text( """ UPDATE document_representations diff --git a/src/rememberstack/workers/__init__.py b/src/rememberstack/workers/__init__.py index cf3dcd2f..e2368566 100644 --- a/src/rememberstack/workers/__init__.py +++ b/src/rememberstack/workers/__init__.py @@ -17,6 +17,9 @@ from rememberstack.workers.e0 import StructurerSettings from rememberstack.workers.e0 import UPLOAD_SOURCE_KIND from rememberstack.workers.e0 import UploadIngestor +from rememberstack.workers.e0_summary import E0_PLACEMENT_VERSION +from rememberstack.workers.e0_summary import E0_SUMMARY_VERSION +from rememberstack.workers.e0_summary import SummarySettings from rememberstack.workers.e1 import ChunkHandler from rememberstack.workers.e1 import E1_CHUNK_VERSION from rememberstack.workers.e1 import E1_EMBED_VERSION @@ -127,6 +130,8 @@ "E0_SKELETON_CHECK_VERSION", "E0_SKELETON_VERSION", "E0_STRUCTURE_VERSION", + "E0_PLACEMENT_VERSION", + "E0_SUMMARY_VERSION", "HandlerOutcome", "HandlerRegistry", "ForgetKnowledgeRebuilder", @@ -157,6 +162,7 @@ "RoleSettings", "SkeletonCheckSettings", "StructurerSettings", + "SummarySettings", "SyncCycleRunner", "SyncSettings", "UPLOAD_SOURCE_KIND", diff --git a/src/rememberstack/workers/e0.py b/src/rememberstack/workers/e0.py index 732d4214..32f6318c 100644 --- a/src/rememberstack/workers/e0.py +++ b/src/rememberstack/workers/e0.py @@ -78,17 +78,19 @@ from rememberstack.ports.object_store import ObjectStorePort from rememberstack.spine.document_catalog import DocumentCatalog from rememberstack.workers.base import HandlerOutcome +from rememberstack.workers.e0_summary import SectionSummarizer +from rememberstack.workers.e0_summary import SummarySettings from rememberstack.workers.e1 import E1_CHUNK_VERSION E0_CONVERT_VERSION: Final = "e0-convert-2026.07" """The convert sub-worker's component version (D12 idempotency key member).""" -E0_STRUCTURE_VERSION: Final = "e0-structure-2026.07d:d79-wave1" -"""The aggregate Wave-1 generation identity. +E0_STRUCTURE_VERSION: Final = "e0-structure-2026.07e:d79-wave2" +"""The aggregate Wave-2 generation identity. Maps old ``e0-structure-2026.07c:temp0-1`` (one-shot offsets/tree/roles/ summaries/placement) to the D79 split: deterministic/anchor skeleton, sanity -check, and role pass. Summary and placement generation slots remain empty. +check, role pass, bottom-up summaries, and root-reduction placement. """ E0_SKELETON_VERSION: Final = ( @@ -538,14 +540,21 @@ def __init__( settings: StructurerSettings | None = None, check_settings: SkeletonCheckSettings | None = None, role_settings: RoleSettings | None = None, + summary_settings: SummarySettings | None = None, ) -> None: - """Bind the handler to all three independently versioned seats.""" + """Bind the handler to all four independently versioned seats.""" self._catalog = catalog self._artifact_store = artifact_store self._model_provider = model_provider self._settings = settings or StructurerSettings() self._check_settings = check_settings or SkeletonCheckSettings() self._role_settings = role_settings or RoleSettings() + self._summarizer = SectionSummarizer( + catalog=catalog, + artifact_store=artifact_store, + model_provider=model_provider, + settings=summary_settings, + ) def handle(self, *, work: ClaimedWork, meter: CostMeterPort) -> HandlerOutcome: """Run the acyclic D79 state machine and flip currency once.""" @@ -559,90 +568,88 @@ def handle(self, *, work: ClaimedWork, meter: CostMeterPort) -> HandlerOutcome: key=ObjectKey(source.markdown_uri) ).decode("utf-8") blocks = blocks_from_sidecar(blocks_doc=blocks_doc, document_md=markdown) - parsed = parse_heading_skeleton( - blocks=blocks, - title=source.title, - markdown_chars=blocks_doc["markdown_chars"], - ) - parsed_analysis = analyze_skeleton( - sections=parsed, blocks=blocks, markdown_chars=len(markdown) - ) - parsed_hash = skeleton_hash(sections=parsed) - - route: StructureRouteTag - selected = parsed - selection_analysis = parsed_analysis - selection_candidate_hash = parsed_hash - selecting_check_id: UUID | None = None - check_version: str | None = None configured_check_version = _skeleton_check_version( settings=self._check_settings ) - producer_family = "N/A" - - if parsed_analysis.heading_density < self._settings.min_heading_density_per_10k: - ( - route, - selected, - selection_analysis, - selection_candidate_hash, - selecting_check_id, - producer_family, - ) = self._fallback_with_terminal( - source=source, - work=work, - blocks=blocks, - markdown=markdown, - meter=meter, - parsed_root=parsed[0], - kept_route=StructureRouteTag.FALLBACK_DENSITY, - ) - check_version = configured_check_version - elif ( - parsed_analysis.oversized_leaf_ratio - > self._settings.max_oversized_leaf_ratio + configured_role_version = _role_version(settings=self._role_settings) + configured_skeleton_version = _skeleton_version(settings=self._settings) + current = self._catalog.current_section_tree( + representation_id=source.representation_id + ) + replay_summary = None + if ( + current is not None + and current.skeleton_version == configured_skeleton_version + and current.skeleton_check_version == configured_check_version + and current.roles_version == configured_role_version ): - ( - route, - selected, - selection_analysis, - selection_candidate_hash, - selecting_check_id, - producer_family, - ) = self._fallback_with_terminal( - source=source, - work=work, - blocks=blocks, - markdown=markdown, - meter=meter, - parsed_root=parsed[0], - kept_route=StructureRouteTag.FALLBACK_LEAF, + # A summary-seat/prompt swap copies the immutable selected + # skeleton + role fields byte-for-byte. Parser/check/role calls + # are skipped; only the new summary/placement generations run. + if ( + current.summary_version == self._summarizer.version + and current.placement_version == self._summarizer.placement_version + and current.placement_path is not None + and all(section.summary is not None for section in current.sections) + ): + selected = current.sections + replay_summary = current + else: + selected = tuple( + section.model_copy(update={"summary": None}) + for section in current.sections + ) + route = current.route_tag + selection_analysis = analyze_skeleton( + sections=selected, blocks=blocks, markdown_chars=len(markdown) ) - check_version = configured_check_version + selection_candidate_hash = current.candidate_skeleton_hash + selecting_check_id = current.selecting_check_id + check_version = current.skeleton_check_version + producer_family = current.skeleton_producer_family else: - initial_outcome, initial_check_id = self._check( - source=source, - work=work, - sections=parsed, - analysis=parsed_analysis, - meter=meter, - terminal=False, + parsed = parse_heading_skeleton( + blocks=blocks, + title=source.title, + markdown_chars=blocks_doc["markdown_chars"], + ) + parsed_analysis = analyze_skeleton( + sections=parsed, blocks=blocks, markdown_chars=len(markdown) ) - selecting_check_id = initial_check_id - check_version = configured_check_version - if _is_incoherent(outcome=initial_outcome): - self._persist_generation( + parsed_hash = skeleton_hash(sections=parsed) + + selected = parsed + selection_analysis = parsed_analysis + selection_candidate_hash = parsed_hash + selecting_check_id = None + check_version = None + producer_family = "N/A" + + if ( + parsed_analysis.heading_density + < self._settings.min_heading_density_per_10k + ): + ( + route, + selected, + selection_analysis, + selection_candidate_hash, + selecting_check_id, + producer_family, + ) = self._fallback_with_terminal( source=source, - sections=parsed, - route=StructureRouteTag.PARSER_DEMOTED_CHECK, - analysis=parsed_analysis, - candidate_skeleton_hash=parsed_hash, - selecting_check_id=initial_check_id, - check_version=configured_check_version, - roles_version=None, - producer_family="N/A", - make_current=False, + work=work, + blocks=blocks, + markdown=markdown, + meter=meter, + parsed_root=parsed[0], + kept_route=StructureRouteTag.FALLBACK_DENSITY, ) + check_version = configured_check_version + elif ( + parsed_analysis.oversized_leaf_ratio + > self._settings.max_oversized_leaf_ratio + ): ( route, selected, @@ -657,21 +664,91 @@ def handle(self, *, work: ClaimedWork, meter: CostMeterPort) -> HandlerOutcome: markdown=markdown, meter=meter, parsed_root=parsed[0], - kept_route=StructureRouteTag.FALLBACK_AFTER_CHECK, + kept_route=StructureRouteTag.FALLBACK_LEAF, ) + check_version = configured_check_version else: - route = StructureRouteTag.PARSER + initial_outcome, initial_check_id = self._check( + source=source, + work=work, + sections=parsed, + analysis=parsed_analysis, + meter=meter, + terminal=False, + ) + selecting_check_id = initial_check_id + check_version = configured_check_version + if _is_incoherent(outcome=initial_outcome): + self._persist_generation( + source=source, + sections=parsed, + route=StructureRouteTag.PARSER_DEMOTED_CHECK, + analysis=parsed_analysis, + candidate_skeleton_hash=parsed_hash, + selecting_check_id=initial_check_id, + check_version=configured_check_version, + roles_version=None, + summary_version=None, + placement_version=None, + placement_path=None, + summary_cache_keys={}, + producer_family="N/A", + make_current=False, + ) + ( + route, + selected, + selection_analysis, + selection_candidate_hash, + selecting_check_id, + producer_family, + ) = self._fallback_with_terminal( + source=source, + work=work, + blocks=blocks, + markdown=markdown, + meter=meter, + parsed_root=parsed[0], + kept_route=StructureRouteTag.FALLBACK_AFTER_CHECK, + ) + else: + route = StructureRouteTag.PARSER - with_roles = self._assign_roles(sections=selected, meter=meter) + selected = self._assign_roles(sections=selected, meter=meter) + + if replay_summary is not None: + replay_placement = replay_summary.placement_path + if replay_placement is None: + raise AssertionError( + "a complete summary generation must carry root placement" + ) + summary_result = self._summarizer.replay( + source=source, + sections=selected, + placement_path=replay_placement, + blocks=blocks, + ) + else: + summary_result = self._summarizer.summarize( + source=source, + sections=selected, + blocks=blocks, + markdown=markdown, + meter=meter, + ) self._persist_generation( source=source, - sections=with_roles, + sections=summary_result.sections, route=route, analysis=selection_analysis, candidate_skeleton_hash=selection_candidate_hash, selecting_check_id=selecting_check_id, check_version=check_version, - roles_version=_role_version(settings=self._role_settings), + roles_version=configured_role_version, + summary_version=summary_result.summary_version, + placement_version=summary_result.placement_version, + placement_path=summary_result.placement_path, + summary_cache_keys=summary_result.cache_keys, producer_family=producer_family, make_current=True, ) @@ -973,6 +1050,10 @@ def _persist_generation( selecting_check_id: UUID | None, check_version: str | None, roles_version: str | None, + summary_version: str | None, + placement_version: str | None, + placement_path: str | None, + summary_cache_keys: dict[str, str], producer_family: str, make_current: bool, ) -> None: @@ -983,16 +1064,24 @@ def _persist_generation( from one seed: a checker bump over an unchanged route+tree derives the same id (ON CONFLICT no-op — no minted generation, no pointer churn), while a genuine route flip or a different fallback tree - changes the seed and appends a real generation. Role version stays - in the seed because roles change the section rows themselves. + changes the seed and appends a real generation. Role, summary, and + placement versions/content stay in the seed because they change the + immutable section rows/sidecar. The checker version remains excluded. """ skeleton_version = _skeleton_version(settings=self._settings) + roles_hash = _roles_hash(sections=sections) + summaries_hash = _summaries_hash(sections=sections) + placement_hash = hashlib.sha256( + (placement_path or "").encode("utf-8") + ).hexdigest() generation_id = uuid5( NAMESPACE_URL, "rememberstack:structure:" f"{source.representation_id}:{E0_STRUCTURE_VERSION}:" f"{route.value}:{skeleton_hash(sections=sections)}:" - f"{skeleton_version}:{roles_version or 'none'}", + f"{skeleton_version}:{roles_version or 'none'}:{roles_hash}:" + f"{summary_version or 'degraded'}:{summaries_hash}:" + f"{placement_version or 'degraded'}:{placement_hash}", ) sidecar_key = ( source.blocks_uri.rsplit("/", 1)[0] @@ -1006,7 +1095,7 @@ def _persist_generation( representation_id=source.representation_id, structure_generation_id=generation_id, sections=sections, - placement_path=None, + placement_path=placement_path, structurer_name=route.value, structurer_version=E0_STRUCTURE_VERSION, skeleton_version=skeleton_version, @@ -1014,6 +1103,8 @@ def _persist_generation( skeleton_producer_family=producer_family, skeleton_check_version=check_version, roles_version=roles_version, + summary_version=summary_version, + placement_version=placement_version, selecting_check_id=selecting_check_id, route_tag=route, candidate_skeleton_hash=candidate_skeleton_hash, @@ -1045,7 +1136,11 @@ def _persist_generation( "stats": persisted.stats.model_dump(mode="json"), "placement": persisted.placement_path, "sections": [ - section.model_dump(mode="json") for section in persisted.sections + { + **section.model_dump(mode="json"), + "summary_cache_key": summary_cache_keys.get(section.node_path), + } + for section in persisted.sections ], } ) @@ -1284,6 +1379,22 @@ def _failure_envelope( } +def _roles_hash(*, sections: tuple[SnappedSection, ...]) -> str: + """Content identity for role-bearing immutable section rows.""" + payload = tuple((section.node_path, section.role) for section in sections) + return hashlib.sha256( + json.dumps(payload, separators=(",", ":")).encode("utf-8") + ).hexdigest() + + +def _summaries_hash(*, sections: tuple[SnappedSection, ...]) -> str: + """Content identity for ordered summary values, including nulls.""" + payload = tuple((section.node_path, section.summary) for section in sections) + return hashlib.sha256( + json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8") + ).hexdigest() + + def _is_incoherent(*, outcome: SkeletonCheckOutcome) -> bool: return outcome.value.startswith("incoherent_") diff --git a/src/rememberstack/workers/e0_summary.py b/src/rememberstack/workers/e0_summary.py new file mode 100644 index 00000000..0567a46b --- /dev/null +++ b/src/rememberstack/workers/e0_summary.py @@ -0,0 +1,823 @@ +"""D79 bottom-up section summaries and root placement. + +Every provider request is bounded before it crosses the port. Leaves and +oversized parent preambles shard only between canonical blocks; wide child +sets reduce through a balanced, versioned fan-in. Independent sections at one +tree depth run in parallel, while the tree itself reduces bottom-up. +""" + +from __future__ import annotations + +from collections import defaultdict +from collections.abc import Iterable +from concurrent.futures import ThreadPoolExecutor +from dataclasses import dataclass +import hashlib +import json +from math import ceil +from typing import Final +from uuid import UUID + +from pydantic import Field +from pydantic_settings import BaseSettings +from pydantic_settings import SettingsConfigDict + +from rememberstack.core import count_tokens +from rememberstack.model import Block +from rememberstack.model import ModelRequest +from rememberstack.model import ObjectKey +from rememberstack.model import ProviderCallError +from rememberstack.model import ProviderCallUsage +from rememberstack.model import RootSummaryPlacementResponse +from rememberstack.model import SectionSummaryResponse +from rememberstack.model import SnappedSection +from rememberstack.model import StructureSource +from rememberstack.ports.cost_meter import CostMeterPort +from rememberstack.ports.model_provider import ModelProviderPort +from rememberstack.ports.object_store import ObjectStorePort +from rememberstack.spine.document_catalog import DocumentCatalog + +SUMMARY_CALL_TOKEN_CEILING: Final = 2_048 +"""Hard per-request ceiling under the pinned whitespace-token counter.""" + +SUMMARY_BALANCED_FAN_IN: Final = 8 +"""Maximum ordered child/partial one-liners in one reduction call.""" + +SUMMARY_MAX_CHARS: Final = 512 +"""Closed response-schema ceiling for every one-line summary.""" + +SUMMARY_PARALLELISM: Final = 8 +"""Maximum independent same-depth section calls in flight.""" + +SUMMARY_TOKEN_COUNTER_VERSION: Final = "whitespace-token-counter-v1" +"""Names ``count_tokens`` semantics in summary-generation provenance.""" + +E0_SUMMARY_VERSION: Final = ( + "e0-summary-2026.07a:d79-bottom-up:block-shard-v1:" + f"{SUMMARY_TOKEN_COUNTER_VERSION}:ceiling{SUMMARY_CALL_TOKEN_CEILING}:" + f"balanced-fan-in{SUMMARY_BALANCED_FAN_IN}:line{SUMMARY_MAX_CHARS}" +) +"""Bottom-up summarizer algorithm and all output-affecting constants.""" + +E0_PLACEMENT_VERSION: Final = "e0-placement-2026.07a:d79-root-reduction-v1" +"""D39 advisory placement emitted by the document-level summary reduction.""" + +_SUMMARY_INSTRUCTION: Final = """Write exactly one factual orientation line \ +of at most 512 characters. Use only the supplied section material. Do not add \ +bullets, labels, markdown, or a newline.""" + +_ROOT_INSTRUCTION: Final = """Return exactly two fields: a one-line factual \ +document summary of at most 512 characters, and one advisory corpus path. The \ +path must begin and end with "/" and contain topical directory names only. \ +Use only the supplied document material.""" + +_REDUCE_INSTRUCTION: Final = """Compose the ordered one-line inputs into \ +exactly one factual orientation line of at most 512 characters. Do not add \ +bullets, labels, markdown, or a newline.""" + +_SECTION_TEMPLATE: Final = """{instruction} +Call kind: section-final +Section path: {node_path} +Section title: {title} +Direct blocks: +{blocks} +Child one-liners: +{children}""" + +_ROOT_TEMPLATE: Final = """{instruction} +Call kind: root-final +Document title: {title} +Source kind: {source_kind} +Direct blocks: +{blocks} +Child one-liners: +{children}""" + +_SHARD_TEMPLATE: Final = """{instruction} +Call kind: block-shard +Section path: {node_path} +Section title: {title} +Shard: {shard_index} +Direct blocks: +{blocks}""" + +_REDUCE_TEMPLATE: Final = """{instruction} +Call kind: {call_kind} +Section path: {node_path} +Section title: {title} +Ordered one-liners: +{lines}""" + +_SUMMARY_PROMPT_HASH: Final = hashlib.sha256( + json.dumps( + { + "instructions": ( + _SUMMARY_INSTRUCTION, + _ROOT_INSTRUCTION, + _REDUCE_INSTRUCTION, + ), + "templates": ( + _SECTION_TEMPLATE, + _ROOT_TEMPLATE, + _SHARD_TEMPLATE, + _REDUCE_TEMPLATE, + ), + "section_schema": SectionSummaryResponse.model_json_schema(), + "root_schema": RootSummaryPlacementResponse.model_json_schema(), + }, + ensure_ascii=False, + separators=(",", ":"), + sort_keys=True, + ).encode("utf-8") +).hexdigest() +"""Hash of every instruction, renderer, and closed response schema.""" + +_GENERATION_PARAMS: Final = { + "temperature": 0.0, + "token_counter": SUMMARY_TOKEN_COUNTER_VERSION, + "token_ceiling": SUMMARY_CALL_TOKEN_CEILING, + "balanced_fan_in": SUMMARY_BALANCED_FAN_IN, + "summary_max_chars": SUMMARY_MAX_CHARS, +} + + +class SummarySettings(BaseSettings): + """The dedicated D70 flash-class D79 summary seat.""" + + model_config = SettingsConfigDict(env_prefix="REMEMBERSTACK_SUMMARY_") + + model: str = Field(default="z-ai/glm-4.7-flash") + + +@dataclass(frozen=True) +class SummaryResult: + """One bottom-up run ready for immutable generation persistence.""" + + sections: tuple[SnappedSection, ...] + placement_path: str | None + summary_version: str | None + placement_version: str | None + cache_keys: dict[str, str] + + +@dataclass(frozen=True) +class _CacheValue: + summary: str + placement_path: str | None + + +@dataclass(frozen=True) +class _MeterEvent: + call_key: str + tier: str + usage: ProviderCallUsage + + +@dataclass(frozen=True) +class _CallResult: + summary: str | None + placement_path: str | None + events: tuple[_MeterEvent, ...] = () + + +class SectionSummarizer: + """Bounded provider orchestration plus sidecar-backed cross-version cache.""" + + def __init__( + self, + *, + catalog: DocumentCatalog, + artifact_store: ObjectStorePort, + model_provider: ModelProviderPort | None, + settings: SummarySettings | None = None, + ) -> None: + """Bind the summary seat to E0's catalog and immutable artifact store.""" + self._catalog = catalog + self._artifact_store = artifact_store + self._model_provider = model_provider + self._settings = settings or SummarySettings() + + @property + def version(self) -> str: + """Hash-stamped summary-seat generation including its D70 model.""" + model_hash = hashlib.sha256(self._settings.model.encode("utf-8")).hexdigest() + return ( + f"{E0_SUMMARY_VERSION}:prompt-{_SUMMARY_PROMPT_HASH[:16]}:" + f"model-{model_hash[:16]}" + ) + + @property + def placement_version(self) -> str: + """Placement is inseparable from this summary root-reduction contract.""" + return f"{E0_PLACEMENT_VERSION}:summary-{hashlib.sha256(self.version.encode()).hexdigest()[:16]}" + + def summarize( + self, + *, + source: StructureSource, + sections: tuple[SnappedSection, ...], + blocks: tuple[Block, ...], + markdown: str, + meter: CostMeterPort, + ) -> SummaryResult: + """Summarize one selected tree bottom-up; every failure is local/null. + + A missing child one-liner makes its ancestors incomplete, so those + ancestors degrade without a partial provider call. Independent + subtrees remain useful and are still persisted. + """ + children = _children_by_parent(sections=sections) + direct = { + section.node_path: _direct_blocks( + section=section, + children=children.get(section.node_path, ()), + blocks=blocks, + ) + for section in sections + } + cache = self._load_cache(doc_id=source.doc_id) + summaries: dict[str, str | None] = {} + placements: dict[str, str | None] = {} + cache_keys: dict[str, str] = {} + + by_depth: defaultdict[int, list[SnappedSection]] = defaultdict(list) + for section in sections: + by_depth[section.node_path.count(".")].append(section) + for depth in sorted(by_depth, reverse=True): + missing: list[tuple[SnappedSection, str, tuple[str, ...]]] = [] + for section in sorted(by_depth[depth], key=lambda item: item.ordinal): + child_sections = children.get(section.node_path, ()) + child_lines = tuple( + summaries.get(child.node_path) for child in child_sections + ) + if any(line is None for line in child_lines): + summaries[section.node_path] = None + placements[section.node_path] = None + continue + complete_child_lines = tuple( + line for line in child_lines if line is not None + ) + key = _summary_cache_key( + section=section, + direct_blocks=direct[section.node_path], + child_summaries=complete_child_lines, + model=self._settings.model, + source_kind=source.source_kind, + ) + cached = cache.get(key) + if cached is not None: + summaries[section.node_path] = cached.summary + placements[section.node_path] = cached.placement_path + cache_keys[section.node_path] = key + continue + missing.append((section, key, complete_child_lines)) + + if not missing: + continue + results: dict[str, _CallResult] = {} + if len(missing) == 1: + section, _, child_lines = missing[0] + results[section.node_path] = self._summarize_uncached( + source=source, + section=section, + direct_blocks=direct[section.node_path], + child_sections=children.get(section.node_path, ()), + child_summaries=child_lines, + markdown=markdown, + ) + else: + with ThreadPoolExecutor( + max_workers=min(SUMMARY_PARALLELISM, len(missing)) + ) as executor: + futures = { + section.node_path: executor.submit( + self._summarize_uncached, + source=source, + section=section, + direct_blocks=direct[section.node_path], + child_sections=children.get(section.node_path, ()), + child_summaries=child_lines, + markdown=markdown, + ) + for section, _, child_lines in missing + } + for path, future in futures.items(): + results[path] = future.result() + + for section, key, _ in missing: + result = results[section.node_path] + for event in result.events: + meter.record( + call_key=event.call_key, tier=event.tier, usage=event.usage + ) + summaries[section.node_path] = result.summary + placements[section.node_path] = result.placement_path + if result.summary is not None: + cache_keys[section.node_path] = key + + summarized = tuple( + section.model_copy(update={"summary": summaries.get(section.node_path)}) + for section in sections + ) + root_summary = summaries.get("0") + placement = placements.get("0") if root_summary is not None else None + complete = all(section.summary is not None for section in summarized) + return SummaryResult( + sections=summarized, + placement_path=placement, + # Provenance convention: a non-null slot certifies a complete + # generation. Partial useful summaries may persist, but a single + # degraded section makes both generation slots null. + summary_version=self.version + if complete and placement is not None + else None, + placement_version=( + self.placement_version if complete and placement is not None else None + ), + cache_keys=cache_keys, + ) + + def replay( + self, + *, + source: StructureSource, + sections: tuple[SnappedSection, ...], + placement_path: str, + blocks: tuple[Block, ...], + ) -> SummaryResult: + """Re-key stored successful summaries without calling the provider. + + This is the retry path when PostgreSQL committed a generation but its + sidecar write did not complete. D7 replays persisted nondeterministic + output; it never asks the model for fresher bytes. + """ + children = _children_by_parent(sections=sections) + direct = { + section.node_path: _direct_blocks( + section=section, + children=children.get(section.node_path, ()), + blocks=blocks, + ) + for section in sections + } + cache_keys: dict[str, str] = {} + summaries = {section.node_path: section.summary for section in sections} + for section in sorted( + sections, key=lambda item: (-item.node_path.count("."), item.ordinal) + ): + summary = summaries[section.node_path] + child_summaries = tuple( + summaries[child.node_path] + for child in children.get(section.node_path, ()) + ) + if summary is None or any(value is None for value in child_summaries): + raise ValueError("only complete summary generations can be replayed") + cache_keys[section.node_path] = _summary_cache_key( + section=section, + direct_blocks=direct[section.node_path], + child_summaries=tuple( + value for value in child_summaries if value is not None + ), + model=self._settings.model, + source_kind=source.source_kind, + ) + return SummaryResult( + sections=sections, + placement_path=placement_path, + summary_version=self.version, + placement_version=self.placement_version, + cache_keys=cache_keys, + ) + + def _summarize_uncached( + self, + *, + source: StructureSource, + section: SnappedSection, + direct_blocks: tuple[Block, ...], + child_sections: tuple[SnappedSection, ...], + child_summaries: tuple[str, ...], + markdown: str, + ) -> _CallResult: + """Run one section's bounded shard/reduce/final call sequence.""" + if self._model_provider is None: + return _CallResult(summary=None, placement_path=None) + + events: list[_MeterEvent] = [] + child_lines = tuple( + f"{child.node_path} | {child.title} | {summary}" + for child, summary in zip(child_sections, child_summaries, strict=True) + ) + reduced_children = self._reduce_lines( + section=section, + lines=child_lines, + call_kind="child-reduction", + call_key_prefix=f"summary:{section.node_path}:children", + events=events, + ) + if reduced_children is None: + return _CallResult(None, None, tuple(events)) + + final_prompt = _render_final_prompt( + source=source, + section=section, + blocks=direct_blocks, + child_lines=reduced_children, + markdown=markdown, + ) + if _within_ceiling(prompt=final_prompt): + result = self._invoke_final( + section=section, prompt=final_prompt, events=events + ) + return result + + shards = _block_shards(section=section, blocks=direct_blocks, markdown=markdown) + if shards is None: + return _CallResult(None, None, tuple(events)) + context_lines: list[str] = [] + for shard_index, shard in enumerate(shards): + prompt = _render_shard_prompt( + section=section, + blocks=shard, + markdown=markdown, + shard_index=shard_index, + ) + partial = self._invoke_summary( + prompt=prompt, + call_key=f"summary:{section.node_path}:shard:{shard_index}", + tier="section_summary_shard", + ) + events.extend(partial.events) + if partial.summary is None: + return _CallResult(None, None, tuple(events)) + context_lines.append(f"direct-shard-{shard_index} | {partial.summary}") + context_lines.extend(reduced_children) + reduced_context = self._reduce_lines( + section=section, + lines=tuple(context_lines), + call_kind="context-reduction", + call_key_prefix=f"summary:{section.node_path}:context", + events=events, + ) + if reduced_context is None: + return _CallResult(None, None, tuple(events)) + final_prompt = _render_final_prompt( + source=source, + section=section, + blocks=(), + child_lines=reduced_context, + markdown=markdown, + ) + if not _within_ceiling(prompt=final_prompt): + return _CallResult(None, None, tuple(events)) + return self._invoke_final(section=section, prompt=final_prompt, events=events) + + def _reduce_lines( + self, + *, + section: SnappedSection, + lines: tuple[str, ...], + call_kind: str, + call_key_prefix: str, + events: list[_MeterEvent], + ) -> tuple[str, ...] | None: + """Balanced fan-in reduction; every intermediate call is bounded.""" + current = lines + level = 0 + while len(current) > SUMMARY_BALANCED_FAN_IN: + reduced: list[str] = [] + for group_index, group in enumerate( + _balanced_groups(values=current, fan_in=SUMMARY_BALANCED_FAN_IN) + ): + prompt = _render_reduce_prompt( + section=section, lines=group, call_kind=call_kind + ) + if not _within_ceiling(prompt=prompt): + return None + result = self._invoke_summary( + prompt=prompt, + call_key=(f"{call_key_prefix}:level:{level}:group:{group_index}"), + tier="section_summary_reduction", + ) + events.extend(result.events) + if result.summary is None: + return None + reduced.append(f"{call_kind}-{level}-{group_index} | {result.summary}") + current = tuple(reduced) + level += 1 + return current + + def _invoke_final( + self, *, section: SnappedSection, prompt: str, events: list[_MeterEvent] + ) -> _CallResult: + """Invoke the one schema that belongs to this section level.""" + if self._model_provider is None: + return _CallResult(None, None, tuple(events)) + call_key = f"summary:{section.node_path}:final" + try: + if section.node_path == "0": + generated = self._model_provider.generate( + request=ModelRequest( + model=self._settings.model, prompt=prompt, temperature=0.0 + ), + response_type=RootSummaryPlacementResponse, + ) + events.append( + _MeterEvent( + call_key=call_key, + tier="document_summary_placement", + usage=generated.usage, + ) + ) + return _CallResult( + generated.output.summary, + generated.output.placement_path, + tuple(events), + ) + generated_summary = self._model_provider.generate( + request=ModelRequest( + model=self._settings.model, prompt=prompt, temperature=0.0 + ), + response_type=SectionSummaryResponse, + ) + events.append( + _MeterEvent( + call_key=call_key, + tier="section_summary", + usage=generated_summary.usage, + ) + ) + return _CallResult(generated_summary.output.summary, None, tuple(events)) + except ProviderCallError as error: + if error.usage is not None: + events.append( + _MeterEvent( + call_key=f"{call_key}:failed", + tier="section_summary_failed_response", + usage=error.usage, + ) + ) + except Exception: # noqa: BLE001 - summary failure never fails a document + pass + return _CallResult(None, None, tuple(events)) + + def _invoke_summary(self, *, prompt: str, call_key: str, tier: str) -> _CallResult: + """Invoke one non-root bounded one-line call.""" + if self._model_provider is None: + return _CallResult(None, None) + try: + generated = self._model_provider.generate( + request=ModelRequest( + model=self._settings.model, prompt=prompt, temperature=0.0 + ), + response_type=SectionSummaryResponse, + ) + return _CallResult( + generated.output.summary, + None, + (_MeterEvent(call_key=call_key, tier=tier, usage=generated.usage),), + ) + except ProviderCallError as error: + if error.usage is not None: + return _CallResult( + None, + None, + ( + _MeterEvent( + call_key=f"{call_key}:failed", + tier=f"{tier}_failed_response", + usage=error.usage, + ), + ), + ) + except Exception: # noqa: BLE001 - summary failure never fails a document + pass + return _CallResult(None, None) + + def _load_cache(self, *, doc_id: UUID) -> dict[str, _CacheValue]: + """Read prior successful generation sidecars; corruption is a cache miss.""" + cache: dict[str, _CacheValue] = {} + for uri in self._catalog.summary_cache_sidecars( + doc_id=doc_id, summary_version=self.version + ): + try: + payload = json.loads( + self._artifact_store.read_bytes(key=ObjectKey(uri)) + ) + placement = payload.get("placement") + for section in payload.get("sections", ()): + key = section.get("summary_cache_key") + summary = section.get("summary") + if not isinstance(key, str) or not isinstance(summary, str): + continue + validated = SectionSummaryResponse(summary=summary) + cached_placement: str | None = None + if section.get("node_path") == "0": + validated_root = RootSummaryPlacementResponse( + summary=validated.summary, placement_path=placement + ) + cached_placement = validated_root.placement_path + cache.setdefault( + key, + _CacheValue( + summary=validated.summary, placement_path=cached_placement + ), + ) + except Exception: # noqa: BLE001 - cache corruption is a miss + # Sidecars are immutable cache metadata, not an availability + # dependency. Missing/legacy/malformed objects simply miss. + continue + return cache + + +def _children_by_parent( + *, sections: tuple[SnappedSection, ...] +) -> dict[str, tuple[SnappedSection, ...]]: + grouped: defaultdict[str, list[SnappedSection]] = defaultdict(list) + for section in sections: + if section.parent_path is not None: + grouped[section.parent_path].append(section) + return { + path: tuple(sorted(values, key=lambda item: item.ordinal)) + for path, values in grouped.items() + } + + +def _direct_blocks( + *, + section: SnappedSection, + children: tuple[SnappedSection, ...], + blocks: tuple[Block, ...], +) -> tuple[Block, ...]: + """Blocks directly owned by a section, excluding parsed heading syntax.""" + excluded: set[int] = set() + for child in children: + excluded.update(range(child.block_start, child.block_end + 1)) + if section.heading_level is not None: + excluded.add(section.block_start) + return tuple( + blocks[index] + for index in range(section.block_start, section.block_end + 1) + if 0 <= index < len(blocks) and index not in excluded + ) + + +def _summary_cache_key( + *, + section: SnappedSection, + direct_blocks: tuple[Block, ...], + child_summaries: tuple[str, ...], + model: str, + source_kind: str, +) -> str: + """The design's exact per-section content/configuration identity.""" + child_hashes = tuple( + hashlib.sha256(summary.encode("utf-8")).hexdigest() + for summary in child_summaries + ) + prompt_hash = hashlib.sha256( + json.dumps( + { + "contract": _SUMMARY_PROMPT_HASH, + "node_path": section.node_path, + "title": section.title, + "source_kind": source_kind if section.node_path == "0" else None, + }, + ensure_ascii=False, + separators=(",", ":"), + sort_keys=True, + ).encode("utf-8") + ).hexdigest() + payload = { + "ordered_constituent_block_hashes": tuple( + block.block_hash for block in direct_blocks + ), + "child_summary_hashes": child_hashes, + "model": model, + "prompt_hash": prompt_hash, + "generation_params": _GENERATION_PARAMS, + "summarizer_version": E0_SUMMARY_VERSION, + } + return hashlib.sha256( + json.dumps( + payload, ensure_ascii=False, separators=(",", ":"), sort_keys=True + ).encode("utf-8") + ).hexdigest() + + +def _render_blocks(*, blocks: Iterable[Block], markdown: str) -> str: + rendered = [ + ( + f"[block ordinal={block.ordinal} type={block.type.value}" + f" hash={block.block_hash}]\n" + f"{markdown[block.char_start : block.char_end]}" + ) + for block in blocks + ] + return "\n\n".join(rendered) if rendered else "(none)" + + +def _render_final_prompt( + *, + source: StructureSource, + section: SnappedSection, + blocks: tuple[Block, ...], + child_lines: tuple[str, ...], + markdown: str, +) -> str: + if section.node_path == "0": + return _ROOT_TEMPLATE.format( + instruction=_ROOT_INSTRUCTION, + title=source.title or "(untitled)", + source_kind=source.source_kind, + blocks=_render_blocks(blocks=blocks, markdown=markdown), + children="\n".join(child_lines) if child_lines else "(none)", + ) + return _SECTION_TEMPLATE.format( + instruction=_SUMMARY_INSTRUCTION, + node_path=section.node_path, + title=section.title or "(untitled section)", + blocks=_render_blocks(blocks=blocks, markdown=markdown), + children="\n".join(child_lines) if child_lines else "(none)", + ) + + +def _render_shard_prompt( + *, + section: SnappedSection, + blocks: tuple[Block, ...], + markdown: str, + shard_index: int, +) -> str: + return _SHARD_TEMPLATE.format( + instruction=_SUMMARY_INSTRUCTION, + node_path=section.node_path, + title=section.title or "(untitled section)", + shard_index=shard_index, + blocks=_render_blocks(blocks=blocks, markdown=markdown), + ) + + +def _render_reduce_prompt( + *, section: SnappedSection, lines: tuple[str, ...], call_kind: str +) -> str: + return _REDUCE_TEMPLATE.format( + instruction=_REDUCE_INSTRUCTION, + call_kind=call_kind, + node_path=section.node_path, + title=section.title or "(untitled section)", + lines="\n".join(lines), + ) + + +def _block_shards( + *, section: SnappedSection, blocks: tuple[Block, ...], markdown: str +) -> tuple[tuple[Block, ...], ...] | None: + """Greedily shard at block grain; an indivisible oversize block degrades.""" + shards: list[tuple[Block, ...]] = [] + current: list[Block] = [] + for block in blocks: + trial = tuple((*current, block)) + prompt = _render_shard_prompt( + section=section, blocks=trial, markdown=markdown, shard_index=len(shards) + ) + if _within_ceiling(prompt=prompt): + current.append(block) + continue + if not current: + return None + shards.append(tuple(current)) + current = [block] + if not _within_ceiling( + prompt=_render_shard_prompt( + section=section, + blocks=tuple(current), + markdown=markdown, + shard_index=len(shards), + ) + ): + return None + if current: + shards.append(tuple(current)) + return tuple(shards) + + +def _balanced_groups( + *, values: tuple[str, ...], fan_in: int +) -> tuple[tuple[str, ...], ...]: + """Partition ordered values into balanced groups no wider than fan-in.""" + if not values: + return () + group_count = ceil(len(values) / fan_in) + small, remainder = divmod(len(values), group_count) + groups: list[tuple[str, ...]] = [] + cursor = 0 + for index in range(group_count): + size = small + (1 if index < remainder else 0) + groups.append(values[cursor : cursor + size]) + cursor += size + return tuple(groups) + + +def _within_ceiling(*, prompt: str) -> bool: + return count_tokens(text=prompt) <= SUMMARY_CALL_TOKEN_CEILING diff --git a/src/tests/profiles/test_selfhost_profile.py b/src/tests/profiles/test_selfhost_profile.py index 11e2542c..bf0b8c8e 100644 --- a/src/tests/profiles/test_selfhost_profile.py +++ b/src/tests/profiles/test_selfhost_profile.py @@ -57,6 +57,20 @@ def test_compose_wires_the_exact_supported_worker_set_and_projection_job() -> No assert 'command: ["project", "--plane", "all"]' in compose +def test_compose_forwards_the_dedicated_summary_seat( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """D66/D70: Compose and readiness expose the same flash-class binding.""" + compose = (_ROOT / "compose.yaml").read_text(encoding="utf-8") + assert ( + "REMEMBERSTACK_SUMMARY_MODEL:" + " ${REMEMBERSTACK_SUMMARY_MODEL:-z-ai/glm-4.7-flash}" + ) in compose + monkeypatch.delenv("REMEMBERSTACK_SUMMARY_MODEL", raising=False) + monkeypatch.setenv("REMEMBERSTACK_OPENROUTER_API_KEY", "test-key") + assert _model_bindings()["section_summary"] == "z-ai/glm-4.7-flash" + + @pytest.mark.parametrize( ("configured", "reported"), (("nebius", "nebius"), ("", "auto")) ) diff --git a/src/tests/spine/test_forget_catalog.py b/src/tests/spine/test_forget_catalog.py index 74dd4be5..955430fb 100644 --- a/src/tests/spine/test_forget_catalog.py +++ b/src/tests/spine/test_forget_catalog.py @@ -200,6 +200,8 @@ def test_inventory_scrub_and_verification_preserve_independent_evidence( "mem://artifacts/target-markdown.md", "mem://artifacts/target-meta.json", "mem://artifacts/target-pageindex.json", + "mem://artifacts/target-structure-old-pageindex.json", + "mem://artifacts/target-structure-new-pageindex.json", "mem://artifacts/writer-transcript.json", f"mem://raw/target.bin?marker={_TOKEN}", "mem://artifacts/planner-transcript.json", @@ -254,11 +256,11 @@ def test_readiness_rehonors_manifest_after_old_postgres_restore( _TARGET_CHECK_ID = UUID("75000000-0000-0000-0000-0000000000c1") _TARGET_GENERATION_ID = UUID("75000000-0000-0000-0000-0000000000c2") +_TARGET_OLD_GENERATION_ID = UUID("75000000-0000-0000-0000-0000000000c3") def _seed_structure_provenance(*, connection: Connection) -> None: - """A D79 check record + generation for the target doc: hard-forget must - erase both EXPLICITLY (review MAJOR — never via implied cascade).""" + """A check plus two immutable generations/sidecars for hard-forget.""" connection.execute( text( "INSERT INTO document_skeleton_checks (" @@ -280,27 +282,35 @@ def _seed_structure_provenance(*, connection: Connection) -> None: "representation": _TARGET_REPRESENTATION_ID, }, ) - connection.execute( - text( - "INSERT INTO document_structure_generations (" - " structure_generation_id, deployment_id, doc_id, version_id," - " representation_id, skeleton_version, skeleton_hash," - " skeleton_producer_family, selecting_check_id, route_tag," - " candidate_skeleton_hash, stats_version, stats" - ") VALUES (" - " :generation_id, :d, :doc, :version, :representation," - " 'skeleton-v', 'hash', 'N/A', :check_id, 'parser', 'hash'," - " 'stats-v', CAST('{}' AS jsonb))" + for generation_id, pageindex_uri in ( + ( + _TARGET_OLD_GENERATION_ID, + "mem://artifacts/target-structure-old-pageindex.json", ), - { - "generation_id": _TARGET_GENERATION_ID, - "d": _DEPLOYMENT_ID, - "doc": _TARGET_DOC_ID, - "version": _TARGET_VERSION_ID, - "representation": _TARGET_REPRESENTATION_ID, - "check_id": _TARGET_CHECK_ID, - }, - ) + (_TARGET_GENERATION_ID, "mem://artifacts/target-structure-new-pageindex.json"), + ): + connection.execute( + text( + "INSERT INTO document_structure_generations (" + " structure_generation_id, deployment_id, doc_id, version_id," + " representation_id, skeleton_version, skeleton_hash," + " skeleton_producer_family, selecting_check_id, route_tag," + " candidate_skeleton_hash, stats_version, stats, pageindex_uri" + ") VALUES (" + " :generation_id, :d, :doc, :version, :representation," + " 'skeleton-v', 'hash', 'N/A', :check_id, 'parser', 'hash'," + " 'stats-v', CAST('{}' AS jsonb), :pageindex_uri)" + ), + { + "generation_id": generation_id, + "d": _DEPLOYMENT_ID, + "doc": _TARGET_DOC_ID, + "version": _TARGET_VERSION_ID, + "representation": _TARGET_REPRESENTATION_ID, + "check_id": _TARGET_CHECK_ID, + "pageindex_uri": pageindex_uri, + }, + ) def _seed_documents(*, connection: Connection) -> None: diff --git a/src/tests/workers/test_d79_structure_route.py b/src/tests/workers/test_d79_structure_route.py index 0aa9d556..4ecb6e2f 100644 --- a/src/tests/workers/test_d79_structure_route.py +++ b/src/tests/workers/test_d79_structure_route.py @@ -52,6 +52,7 @@ class _Catalog: def __init__(self) -> None: self.checks: list[SkeletonCheckRecord] = [] self.generations: list[SectionTreeRecord] = [] + self.current: PersistedSectionTree | None = None def structure_source(self, *, representation_id: UUID) -> StructureSource: assert representation_id == _REPRESENTATION @@ -69,9 +70,22 @@ def structure_source(self, *, representation_id: UUID) -> StructureSource: def record_skeleton_check(self, *, record: SkeletonCheckRecord) -> None: self.checks.append(record) + def current_section_tree( + self, *, representation_id: UUID + ) -> PersistedSectionTree | None: + assert representation_id == _REPRESENTATION + return self.current + + def summary_cache_sidecars( + self, *, doc_id: UUID, summary_version: str + ) -> tuple[str, ...]: + assert doc_id == _DOC + assert summary_version + return () + def record_section_tree(self, *, record: SectionTreeRecord) -> PersistedSectionTree: self.generations.append(record) - return PersistedSectionTree( + persisted = PersistedSectionTree( sections=record.sections, placement_path=record.placement_path, structurer_version=record.structurer_version, @@ -90,6 +104,9 @@ def record_section_tree(self, *, record: SectionTreeRecord) -> PersistedSectionT stats_version=record.stats_version, stats=record.stats, ) + if record.make_current: + self.current = persisted + return persisted def _work() -> ClaimedWork: @@ -125,6 +142,13 @@ def route(prompt: str, response_type: str) -> dict[str, object]: } if response_type == "RoleClassificationResponse": return {"assignments": []} + if response_type == "SectionSummaryResponse": + return {"summary": "A bounded section summary."} + if response_type == "RootSummaryPlacementResponse": + return { + "summary": "A bounded document summary.", + "placement_path": "/routing/proof/", + } raise AssertionError(response_type) return route @@ -423,11 +447,22 @@ def test_checker_bump_does_not_mint_a_new_generation_when_route_holds( ).encode(), ) catalog = _Catalog() + provider = FakeModelProvider( + generate_router=lambda prompt, response_type: ( + {"verdict": "coherent"} + if response_type == "SkeletonCheckResponse" + else {"assignments": []} + if response_type == "RoleClassificationResponse" + else {"summary": "Stable one-line summary."} + if response_type == "SectionSummaryResponse" + else {"summary": "Stable root summary.", "placement_path": "/stable/topic/"} + ) + ) for checker_model in ("checker/generation-one", "checker/generation-two"): StructureHandler( catalog=catalog, # type: ignore[arg-type] artifact_store=store, - model_provider=None, + model_provider=provider, settings=StructurerSettings( min_heading_density_per_10k=0, max_oversized_leaf_ratio=1 ), @@ -439,6 +474,8 @@ def test_checker_bump_does_not_mint_a_new_generation_when_route_holds( len({record.structure_generation_id for record in catalog.generations}) == 1 ) # ...but they derive the SAME id: ON CONFLICT makes the second a no-op assert len(catalog.checks) == 2 # while every check outcome is ledgered + assert all(record.summary_version is not None for record in catalog.generations) + assert all(record.placement_version is not None for record in catalog.generations) def test_role_classifier_failure_defaults_to_body(tmp_path: Path) -> None: diff --git a/src/tests/workers/test_d79_summaries.py b/src/tests/workers/test_d79_summaries.py new file mode 100644 index 00000000..4314b5ef --- /dev/null +++ b/src/tests/workers/test_d79_summaries.py @@ -0,0 +1,220 @@ +"""D79 bottom-up summary composition and hard request bounds.""" + +from pathlib import Path +from uuid import UUID + +from rememberstack.adapters.selfhost import LocalFSObjectStore +from rememberstack.adapters.testing import FakeModelProvider +from rememberstack.adapters.testing import NoopCostMeter +from rememberstack.core import blockize +from rememberstack.core import count_tokens +from rememberstack.core import parse_heading_skeleton +from rememberstack.model import RootSummaryPlacementResponse +from rememberstack.model import SectionSummaryResponse +from rememberstack.model import StructureSource +from rememberstack.workers.e0_summary import SectionSummarizer +from rememberstack.workers.e0_summary import SUMMARY_BALANCED_FAN_IN +from rememberstack.workers.e0_summary import SUMMARY_CALL_TOKEN_CEILING +from rememberstack.workers.e0_summary import SummarySettings + +_DEPLOYMENT = UUID("71000000-0000-0000-0000-000000000001") +_DOC = UUID("71000000-0000-0000-0000-000000000002") +_VERSION = UUID("71000000-0000-0000-0000-000000000003") +_REPRESENTATION = UUID("71000000-0000-0000-0000-000000000004") + + +class _NoCacheCatalog: + """The unit proofs exercise composition without prior sidecars.""" + + def summary_cache_sidecars( + self, *, doc_id: UUID, summary_version: str + ) -> tuple[str, ...]: + assert doc_id == _DOC + assert summary_version + return () + + +def _source() -> StructureSource: + return StructureSource( + deployment_id=_DEPLOYMENT, + doc_id=_DOC, + version_id=_VERSION, + representation_id=_REPRESENTATION, + blocks_uri="unused/blocks.json", + markdown_uri="unused/document.md", + title="Composition proof", + source_kind="upload", + ) + + +def _summarize(*, tmp_path: Path, markdown: str, provider: FakeModelProvider): + blocks = blockize(document_md=markdown) + sections = parse_heading_skeleton( + blocks=blocks, title="Composition proof", markdown_chars=len(markdown) + ) + result = SectionSummarizer( + catalog=_NoCacheCatalog(), # type: ignore[arg-type] + artifact_store=LocalFSObjectStore(root=tmp_path), + model_provider=provider, + settings=SummarySettings(model="summary/test"), + ).summarize( + source=_source(), + sections=sections, + blocks=blocks, + markdown=markdown, + meter=NoopCostMeter(), + ) + return result, provider.generated_requests + + +def test_leaf_parent_root_read_exactly_their_composition_inputs(tmp_path: Path) -> None: + markdown = "\n\n".join( + ( + "ROOT-OWN preamble.", + "# Parent", + "PARENT-OWN preamble.", + "## Leaf Alpha", + "ALPHA-ONLY body.", + "## Leaf Beta", + "BETA-ONLY body.", + ) + ) + + def route(prompt: str, response_type: str) -> dict[str, object]: + if response_type == "RootSummaryPlacementResponse": + assert "ROOT-OWN preamble." in prompt + assert "Parent composed line." in prompt + assert "PARENT-OWN preamble." not in prompt + assert "ALPHA-ONLY body." not in prompt + return { + "summary": "Document composed line.", + "placement_path": "/proof/composition/", + } + assert response_type == "SectionSummaryResponse" + if "Call kind: section-final\nSection path: 0.0.0\n" in prompt: + assert "ALPHA-ONLY body." in prompt + assert "BETA-ONLY body." not in prompt + assert "PARENT-OWN preamble." not in prompt + return {"summary": "Alpha leaf line."} + if "Call kind: section-final\nSection path: 0.0.1\n" in prompt: + assert "BETA-ONLY body." in prompt + assert "ALPHA-ONLY body." not in prompt + assert "PARENT-OWN preamble." not in prompt + return {"summary": "Beta leaf line."} + assert "Call kind: section-final\nSection path: 0.0\n" in prompt + assert "PARENT-OWN preamble." in prompt + assert "Alpha leaf line." in prompt + assert "Beta leaf line." in prompt + assert "ALPHA-ONLY body." not in prompt + assert "BETA-ONLY body." not in prompt + return {"summary": "Parent composed line."} + + result, requests = _summarize( + tmp_path=tmp_path, + markdown=markdown, + provider=FakeModelProvider(generate_router=route), + ) + + assert [section.summary for section in result.sections] == [ + "Document composed line.", + "Parent composed line.", + "Alpha leaf line.", + "Beta leaf line.", + ] + assert result.placement_path == "/proof/composition/" + assert result.summary_version is not None + assert result.placement_version is not None + assert all(request.temperature == 0.0 for request in requests) + + +def test_oversized_leaf_shards_at_block_grain_then_composes(tmp_path: Path) -> None: + paragraphs = [ + f"BLOCK-{index} " + " ".join(f"word-{index}-{word}" for word in range(90)) + for index in range(80) + ] + markdown = "\n\n".join(("# Giant", *paragraphs)) + + def route(prompt: str, response_type: str) -> dict[str, object]: + if response_type == "RootSummaryPlacementResponse": + return {"summary": "A giant document.", "placement_path": "/proof/giant/"} + assert response_type == "SectionSummaryResponse" + if "Call kind: block-shard" in prompt: + return {"summary": "One bounded shard."} + return {"summary": "The giant section composed from bounded shards."} + + result, requests = _summarize( + tmp_path=tmp_path, + markdown=markdown, + provider=FakeModelProvider(generate_router=route), + ) + + shard_prompts = [ + request.prompt + for request in requests + if "Call kind: block-shard" in request.prompt + ] + assert len(shard_prompts) > 1 + assert all( + count_tokens(text=request.prompt) <= SUMMARY_CALL_TOKEN_CEILING + for request in requests + ) + assert result.sections[1].summary == ( + "The giant section composed from bounded shards." + ) + assert result.sections[0].summary == "A giant document." + + +def test_many_children_use_balanced_bounded_fan_in(tmp_path: Path) -> None: + markdown = "\n\n".join( + ( + "# Parent", + "Parent preamble.", + *( + value + for index in range(18) + for value in (f"## Child {index}", f"Body {index}.") + ), + ) + ) + + def route(prompt: str, response_type: str) -> dict[str, object]: + if response_type == "RootSummaryPlacementResponse": + return {"summary": "Wide document.", "placement_path": "/proof/wide/"} + if "Call kind: child-reduction" in prompt: + return {"summary": "Balanced child group."} + if "Section path: 0.0\n" in prompt: + return {"summary": "Wide parent."} + return {"summary": "Child line."} + + result, requests = _summarize( + tmp_path=tmp_path, + markdown=markdown, + provider=FakeModelProvider(generate_router=route), + ) + reductions = [ + request.prompt + for request in requests + if "Call kind: child-reduction" in request.prompt + and "Section path: 0.0\n" in request.prompt + ] + assert len(reductions) == 3 + assert { + len(prompt.split("Ordered one-liners:\n", 1)[1].splitlines()) + for prompt in reductions + } == {6} + assert all( + len(prompt.split("Ordered one-liners:\n", 1)[1].splitlines()) + <= SUMMARY_BALANCED_FAN_IN + for prompt in reductions + ) + assert result.sections[1].summary == "Wide parent." + + +def test_summary_response_schemas_are_closed_and_bounded() -> None: + section_schema = SectionSummaryResponse.model_json_schema() + root_schema = RootSummaryPlacementResponse.model_json_schema() + assert section_schema["additionalProperties"] is False + assert root_schema["additionalProperties"] is False + assert section_schema["properties"]["summary"]["maxLength"] == 512 + assert root_schema["properties"]["summary"]["maxLength"] == 512 + assert root_schema["properties"]["placement_path"]["maxLength"] == 512 diff --git a/src/tests/workers/test_e0_chain.py b/src/tests/workers/test_e0_chain.py index e285be13..5766b8fc 100644 --- a/src/tests/workers/test_e0_chain.py +++ b/src/tests/workers/test_e0_chain.py @@ -41,14 +41,17 @@ from rememberstack.spine import DeploymentBootstrapper from rememberstack.spine import DocumentCatalog from rememberstack.spine import ForgetCatalog +from rememberstack.spine import ProjectionCatalog from rememberstack.spine import WorkLedger from rememberstack.spine import WorkLedgerSettings from rememberstack.spine.settings import load_database_settings from rememberstack.workers import ConvertHandler from rememberstack.workers import E0_CONVERT_VERSION +from rememberstack.workers import E0_STRUCTURE_VERSION from rememberstack.workers import HandlerRegistry from rememberstack.workers import StructureHandler from rememberstack.workers import StructurerSettings +from rememberstack.workers import SummarySettings from rememberstack.workers import UploadIngestor from rememberstack.workers import Worker @@ -484,7 +487,9 @@ def test_empty_document_gets_an_empty_root_span(rig: _E0Rig) -> None: ) -def _structure_worker(rig: _E0Rig, provider: object) -> Worker: +def _structure_worker( + rig: _E0Rig, provider: object, *, summary_model: str = "summary/test" +) -> Worker: """A worker whose structure stage runs the full LLM route.""" registry = HandlerRegistry() registry.register( @@ -494,17 +499,25 @@ def _structure_worker(rig: _E0Rig, provider: object) -> Worker: artifact_store=rig.artifact_store, model_provider=provider, # type: ignore[arg-type] settings=StructurerSettings(min_blocks_for_llm=3), + summary_settings=SummarySettings(model=summary_model), ), ) return Worker(ledger=rig.ledger, registry=registry) -def test_full_structure_route_persists_the_snapped_tree(rig: _E0Rig) -> None: - """D79 parser rows, deterministic roles, null placement, versioned sidecar.""" +def test_full_structure_route_persists_summaries_and_root_placement( + rig: _E0Rig, +) -> None: + """D79 parser rows, summaries, placement, sidecar, and P3 source query.""" provider = FakeModelProvider( generate_payloads={ "SkeletonCheckResponse": {"verdict": "coherent"}, "RoleClassificationResponse": {"assignments": []}, + "SectionSummaryResponse": {"summary": "A section orientation line."}, + "RootSummaryPlacementResponse": { + "summary": "A field report about erosion.", + "placement_path": "/field-research/erosion/", + }, } ) ingested = rig.ingestor.ingest( @@ -529,7 +542,8 @@ def test_full_structure_route_persists_the_snapped_tree(rig: _E0Rig) -> None: connection.execute( text( "SELECT node_path, parent_section_id, section_id, title," - " role::text AS role, block_start, block_end, placement_path" + " role::text AS role, block_start, block_end, summary," + " placement_path" " FROM document_sections WHERE version_id = :version_id" " ORDER BY ordinal" ), @@ -547,7 +561,8 @@ def test_full_structure_route_persists_the_snapped_tree(rig: _E0Rig) -> None: ).scalar_one() assert [row["node_path"] for row in sections] == ["0", "0.0", "0.0.0", "0.0.1"] root, report, findings, recommendations = sections - assert root["placement_path"] is None + assert root["summary"] == "A field report about erosion." + assert root["placement_path"] == "/field-research/erosion/" assert report["parent_section_id"] == root["section_id"] assert findings["parent_section_id"] == report["section_id"] assert recommendations["parent_section_id"] == report["section_id"] @@ -566,10 +581,18 @@ def test_full_structure_route_persists_the_snapped_tree(rig: _E0Rig) -> None: key=ObjectKey(str(representation["pageindex_uri"])) ) ) - assert sidecar["placement"] is None - assert sidecar["generations"]["summary"] is None - assert sidecar["generations"]["placement"] is None + assert sidecar["placement"] == "/field-research/erosion/" + assert sidecar["generations"]["summary"] is not None + assert sidecar["generations"]["placement"] is not None + assert all(section["summary"] for section in sidecar["sections"]) + assert all(section["summary_cache_key"] for section in sidecar["sections"]) assert len(sidecar["sections"]) == 4 + # This is the exact current-generation query consumed by CorpusFsBuilder. + p3_document = ProjectionCatalog(engine=rig.engine).corpus_documents( + deployment_id=_DEPLOYMENT_ID + )[0] + assert p3_document["root_summary"] == "A field report about erosion." + assert p3_document["placement_path"] == "/field-research/erosion/" def test_failed_checker_is_explicit_and_fail_open_on_the_parser(rig: _E0Rig) -> None: @@ -623,6 +646,307 @@ def embed(self, *, request: object) -> object: assert len(checks) == 1 assert checks[0].check_outcome == "provider_error" assert checks[0].provider_failure is not None + degraded = rig.row( + sql="SELECT g.summary_version, g.placement_version, s.summary," + " s.placement_path FROM document_representations r" + " JOIN document_structure_generations g" + " ON g.structure_generation_id = r.current_structure_generation_id" + " JOIN document_sections s" + " ON s.structure_generation_id = g.structure_generation_id" + " AND s.node_path = '0'" + " WHERE r.representation_id = :representation_id", + params={"representation_id": representation["representation_id"]}, + ) + assert degraded == { + "summary_version": None, + "placement_version": None, + "summary": None, + "placement_path": None, + } + + +def test_summary_cache_recomputes_only_edited_leaf_and_ancestors(rig: _E0Rig) -> None: + """Two lineage versions: one edited leaf misses; its sibling cache-hits.""" + summary_calls: list[str] = [] + + def route(prompt: str, response_type: str) -> dict[str, object]: + if response_type == "SkeletonCheckResponse": + return {"verdict": "coherent"} + if response_type == "RoleClassificationResponse": + return {"assignments": []} + path = ( + "0" + if response_type == "RootSummaryPlacementResponse" + else prompt.split("Section path: ", 1)[1].splitlines()[0] + ) + summary_calls.append(path) + edited = "EDITED" in prompt or "revised" in prompt + if response_type == "RootSummaryPlacementResponse": + return { + "summary": ( + "Revised field report." if edited else "Original field report." + ), + "placement_path": "/field-research/erosion/", + } + if path == "0.0.1": + return { + "summary": ( + "Recommendations revised." + if edited + else "Original recommendations." + ) + } + if path == "0.0": + return { + "summary": ( + "Field report revised." if edited else "Original field report." + ) + } + return {"summary": "Stable findings."} + + provider = FakeModelProvider(generate_router=route) + worker = _structure_worker(rig, provider, summary_model="summary/cache-proof") + + first = rig.ingestor.ingest_observed( + deployment_id=_DEPLOYMENT_ID, + source_kind="watched_directory", + source_ref="reports/erosion.md", + upload=DocumentUpload( + filename="erosion.md", + mime="text/markdown", + content=_STRUCTURED_SOURCE.encode("utf-8"), + ), + versioning_mode="living", + source_modified_at=None, + source_version_ref="v1", + sync_cycle_id=None, + ) + assert rig.run(stage=PipelineStage.CONVERT) is RunResultOutcome.SUCCEEDED + assert ( + worker.run_one( + deployment_id=_DEPLOYMENT_ID, + stage=PipelineStage.STRUCTURE, + lane=ProcessingLane.STEADY, + ).outcome + is RunResultOutcome.SUCCEEDED + ) + assert set(summary_calls[:2]) == {"0.0.0", "0.0.1"} + assert summary_calls[2:] == ["0.0", "0"] + + summary_calls.clear() + edited_source = _STRUCTURED_SOURCE.replace( + "Publish the dataset.", "Publish the EDITED dataset." + ) + second = rig.ingestor.ingest_observed( + deployment_id=_DEPLOYMENT_ID, + source_kind="watched_directory", + source_ref="reports/erosion.md", + upload=DocumentUpload( + filename="erosion.md", + mime="text/markdown", + content=edited_source.encode("utf-8"), + ), + versioning_mode="living", + source_modified_at=None, + source_version_ref="v2", + sync_cycle_id=None, + ) + assert second.doc_id == first.doc_id + assert second.version_id != first.version_id + assert rig.run(stage=PipelineStage.CONVERT) is RunResultOutcome.SUCCEEDED + assert ( + worker.run_one( + deployment_id=_DEPLOYMENT_ID, + stage=PipelineStage.STRUCTURE, + lane=ProcessingLane.STEADY, + ).outcome + is RunResultOutcome.SUCCEEDED + ) + + assert summary_calls == ["0.0.1", "0.0", "0"] + rows = rig.row( + sql="SELECT count(*) AS generations," + " count(*) FILTER (WHERE summary_version IS NOT NULL) AS complete" + " FROM document_structure_generations WHERE doc_id = :doc_id", + params={"doc_id": first.doc_id}, + ) + assert rows == {"generations": 2, "complete": 2} + + +def test_summary_seat_swap_copies_skeleton_and_moves_pointer( + rig: _E0Rig, monkeypatch: pytest.MonkeyPatch +) -> None: + """A D70 swap mints summary/placement only; parser/check/roles are replayed.""" + + def provider(marker: str) -> FakeModelProvider: + def route(prompt: str, response_type: str) -> dict[str, object]: + if response_type == "SkeletonCheckResponse": + return {"verdict": "coherent"} + if response_type == "RoleClassificationResponse": + return {"assignments": []} + if response_type == "RootSummaryPlacementResponse": + return { + "summary": f"{marker} root summary.", + "placement_path": f"/field-research/{marker}/", + } + path = prompt.split("Section path: ", 1)[1].splitlines()[0] + return {"summary": f"{marker} summary for {path}."} + + return FakeModelProvider(generate_router=route) + + first_provider = provider("alpha") + ingested = rig.ingestor.ingest( + deployment_id=_DEPLOYMENT_ID, + upload=DocumentUpload( + filename="survey.md", + mime="text/markdown", + content=_STRUCTURED_SOURCE.encode("utf-8"), + ), + ) + assert rig.run(stage=PipelineStage.CONVERT) is RunResultOutcome.SUCCEEDED + first_worker = _structure_worker( + rig, first_provider, summary_model="summary/model-alpha" + ) + assert ( + first_worker.run_one( + deployment_id=_DEPLOYMENT_ID, + stage=PipelineStage.STRUCTURE, + lane=ProcessingLane.STEADY, + ).outcome + is RunResultOutcome.SUCCEEDED + ) + representation = rig.row( + sql="SELECT representation_id, current_structure_generation_id" + " FROM document_representations WHERE version_id = :version_id", + params={"version_id": ingested.version_id}, + ) + first_generation = representation["current_structure_generation_id"] + + def parser_must_not_run(**kwargs: object) -> object: + del kwargs + raise AssertionError("summary-seat swap re-parsed the representation") + + monkeypatch.setattr( + "rememberstack.workers.e0.parse_heading_skeleton", parser_must_not_run + ) + second_provider = provider("beta") + StructureHandler( + catalog=rig.catalog, + artifact_store=rig.artifact_store, + model_provider=second_provider, + settings=StructurerSettings(min_blocks_for_llm=3), + summary_settings=SummarySettings(model="summary/model-beta"), + ).handle( + work=ClaimedWork( + processing_id=uuid4(), + deployment_id=_DEPLOYMENT_ID, + target_kind=ProcessingTarget.DOCUMENT_VERSION, + target_id=ingested.version_id, + stage=PipelineStage.STRUCTURE, + component_version=E0_STRUCTURE_VERSION, + content_hash=ingested.content_hash, + lane=ProcessingLane.STEADY, + attempt=1, + payload={ + "version_id": str(ingested.version_id), + "representation_id": str(representation["representation_id"]), + }, + ), + meter=NoopCostMeter(), + ) + + with rig.engine.connect() as connection: + generations = ( + connection.execute( + text( + "SELECT structure_generation_id, skeleton_version," + " skeleton_hash, skeleton_producer_family," + " skeleton_check_version, roles_version, summary_version," + " placement_version, selecting_check_id, route_tag::text," + " candidate_skeleton_hash, stats_version, stats" + " FROM document_structure_generations" + " WHERE representation_id = :representation_id" + " ORDER BY created_at, structure_generation_id" + ), + {"representation_id": representation["representation_id"]}, + ) + .mappings() + .all() + ) + section_rows = ( + connection.execute( + text( + "SELECT structure_generation_id, node_path, title," + " role::text AS role, block_start, block_end, char_start," + " char_end, ordinal, heading_level, normalized_title, summary," + " placement_path FROM document_sections" + " WHERE representation_id = :representation_id" + " ORDER BY structure_generation_id, ordinal" + ), + {"representation_id": representation["representation_id"]}, + ) + .mappings() + .all() + ) + current = connection.execute( + text( + "SELECT current_structure_generation_id" + " FROM document_representations" + " WHERE representation_id = :representation_id" + ), + {"representation_id": representation["representation_id"]}, + ).scalar_one() + + assert len(generations) == 2 + assert generations[0]["structure_generation_id"] == first_generation + assert generations[1]["structure_generation_id"] == current + assert current != first_generation + unchanged_generation_fields = ( + "skeleton_version", + "skeleton_hash", + "skeleton_producer_family", + "skeleton_check_version", + "roles_version", + "selecting_check_id", + "route_tag", + "candidate_skeleton_hash", + "stats_version", + "stats", + ) + assert {key: generations[0][key] for key in unchanged_generation_fields} == { + key: generations[1][key] for key in unchanged_generation_fields + } + assert generations[0]["summary_version"] != generations[1]["summary_version"] + assert generations[0]["placement_version"] != generations[1]["placement_version"] + + by_generation: dict[object, list[dict[str, object]]] = {} + for row in section_rows: + by_generation.setdefault(row["structure_generation_id"], []).append(dict(row)) + first_sections = by_generation[first_generation] + second_sections = by_generation[current] + skeleton_role_fields = ( + "node_path", + "title", + "role", + "block_start", + "block_end", + "char_start", + "char_end", + "ordinal", + "heading_level", + "normalized_title", + ) + assert [ + {key: row[key] for key in skeleton_role_fields} for row in first_sections + ] == [{key: row[key] for key in skeleton_role_fields} for row in second_sections] + assert {str(row["summary"]) for row in first_sections} != { + str(row["summary"]) for row in second_sections + } + assert second_sections[0]["placement_path"] == "/field-research/beta/" + assert not any( + request.model != "summary/model-beta" + for request in second_provider.generated_requests + ) def test_retried_tree_write_returns_the_first_attempts_truth(rig: _E0Rig) -> None: diff --git a/website/src/app/docs/deployment/page.mdx b/website/src/app/docs/deployment/page.mdx index df569a67..22765aaa 100644 --- a/website/src/app/docs/deployment/page.mdx +++ b/website/src/app/docs/deployment/page.mdx @@ -17,13 +17,15 @@ The RememberStack, `rememberstack`, and `remember` names used here are final. Do not expose this stack to untrusted traffic. Model seats are per-deployment environment bindings. The document-structure -stage uses three: `REMEMBERSTACK_SKELETON_CHECK_MODEL` (the bounded skeleton +stage uses four: `REMEMBERSTACK_SKELETON_CHECK_MODEL` (the bounded skeleton sanity judge, default `z-ai/glm-4.7-flash`), `REMEMBERSTACK_ROLE_MODEL` (the -title-only section-role classifier, default `z-ai/glm-4.7-flash`), and -`REMEMBERSTACK_STRUCTURER_MODEL` — which now names **only** the string-anchor +title-only section-role classifier, default `z-ai/glm-4.7-flash`), +`REMEMBERSTACK_SUMMARY_MODEL` (bottom-up section summaries plus the root +placement reduction, default `z-ai/glm-4.7-flash`), and +`REMEMBERSTACK_STRUCTURER_MODEL` — which names **only** the string-anchor fallback proposer for documents whose headings cannot be parsed or whose -parsed tree fails the sanity check; the routine path is deterministic and -calls no model. +parsed tree fails the sanity check. Routine skeleton construction is +deterministic; bounded summary calls still run on the selected tree. ## Start the stack From 6239447a74b67913dc97658023fadf811dae3389 Mon Sep 17 00:00:00 2001 From: Jiri Puc Date: Wed, 29 Jul 2026 10:04:44 +0200 Subject: [PATCH 2/2] fix(e0): apply D79 Wave-2 review findings --- plan/designs/e0_files_design.md | 11 +- src/rememberstack/model/sections.py | 8 +- src/rememberstack/spine/document_catalog.py | 10 +- src/rememberstack/workers/e0.py | 124 +++--- src/rememberstack/workers/e0_summary.py | 248 ++++++++---- src/tests/workers/test_d79_structure_route.py | 68 ++-- src/tests/workers/test_d79_summaries.py | 257 ++++++++++++- src/tests/workers/test_e0_chain.py | 362 ++++++++++++++++++ website/src/app/docs/deployment/page.mdx | 4 +- 9 files changed, 913 insertions(+), 179 deletions(-) diff --git a/plan/designs/e0_files_design.md b/plan/designs/e0_files_design.md index dcd71ab3..9a837f46 100644 --- a/plan/designs/e0_files_design.md +++ b/plan/designs/e0_files_design.md @@ -382,10 +382,13 @@ unchanged): bounded: a leaf larger than the summary-call token ceiling is sharded at block grain and reduced; a parent call reads its **own direct blocks plus its children's one-liners** (a chapter's preamble is content too) with balanced fan-in when children are many. Calls are - parallel and cached per section; the cache key is the ordered constituent block hashes, - child-summary hashes, model, prompt, generation parameters, and summarizer version. This is - the call shape where cheap models are reliable: bounded context, no arithmetic, no giant - output. **Degradation convention:** `summary_version` is non-null only when every section + parallel and cached per section; the cache key hashes the ordered rendered block strings, + rendered child lines (path + capped title + summary), model, prompt, generation parameters, + and summarizer version. This is the call shape where cheap models are reliable: bounded + context, no arithmetic, no giant output. The versioned token estimate is whitespace-based; + the companion character ceiling is the hard bound for whitespace-poor/CJK/minified input, + and every rendered call must fit both. **Degradation convention:** `summary_version` is + non-null only when every section in the generation has a summary; any failed section call leaves that section null (and any ancestor that cannot receive all child one-liners null), may preserve successful summaries in independent subtrees, and makes the generation's `summary_version` null. Because root diff --git a/src/rememberstack/model/sections.py b/src/rememberstack/model/sections.py index c78a89ee..aba97cd1 100644 --- a/src/rememberstack/model/sections.py +++ b/src/rememberstack/model/sections.py @@ -165,11 +165,11 @@ class RoleClassificationResponse(BaseModel): class SectionSummaryResponse(BaseModel): - """One bounded, single-line section summary and no auxiliary prose.""" + """One section summary and no auxiliary prose.""" model_config = ConfigDict(frozen=True, extra="forbid") - summary: str = Field(min_length=1, max_length=512, pattern=r"^[^\r\n]+$") + summary: str = Field(min_length=1) class RootSummaryPlacementResponse(BaseModel): @@ -177,8 +177,8 @@ class RootSummaryPlacementResponse(BaseModel): model_config = ConfigDict(frozen=True, extra="forbid") - summary: str = Field(min_length=1, max_length=512, pattern=r"^[^\r\n]+$") - placement_path: str = Field(min_length=3, max_length=512, pattern=r"^/[^\r\n]+/$") + summary: str = Field(min_length=1) + placement_path: str = Field(min_length=1) class SnappedSection(BaseModel): diff --git a/src/rememberstack/spine/document_catalog.py b/src/rememberstack/spine/document_catalog.py index 365f635a..053a85cd 100644 --- a/src/rememberstack/spine/document_catalog.py +++ b/src/rememberstack/spine/document_catalog.py @@ -244,14 +244,11 @@ def current_section_tree( raise RuntimeError("current structure generation has no root section") return _persisted_tree(generation=generation, sections=persisted) - def summary_cache_sidecars( - self, *, doc_id: UUID, summary_version: str - ) -> tuple[str, ...]: - """Prior successful summary generations for sidecar cache lookup.""" + def summary_cache_sidecars(self, *, doc_id: UUID) -> tuple[str, ...]: + """All prior sidecars that may contain keyed summary cache entries.""" with self._engine.connect() as connection: rows = connection.execute( - _SELECT_SUMMARY_CACHE_SIDECARS, - {"doc_id": doc_id, "summary_version": summary_version}, + _SELECT_SUMMARY_CACHE_SIDECARS, {"doc_id": doc_id} ).scalars() return tuple(str(uri) for uri in rows) @@ -754,7 +751,6 @@ def _lineage_locked(*, connection: Connection, record: UploadRecord) -> UUID: SELECT pageindex_uri FROM document_structure_generations WHERE doc_id = :doc_id - AND summary_version = :summary_version AND pageindex_uri IS NOT NULL ORDER BY created_at DESC, structure_generation_id """ diff --git a/src/rememberstack/workers/e0.py b/src/rememberstack/workers/e0.py index 32f6318c..e6b01b73 100644 --- a/src/rememberstack/workers/e0.py +++ b/src/rememberstack/workers/e0.py @@ -85,12 +85,13 @@ E0_CONVERT_VERSION: Final = "e0-convert-2026.07" """The convert sub-worker's component version (D12 idempotency key member).""" -E0_STRUCTURE_VERSION: Final = "e0-structure-2026.07e:d79-wave2" +E0_STRUCTURE_VERSION: Final = "e0-structure-2026.07f:d79-wave2" """The aggregate Wave-2 generation identity. Maps old ``e0-structure-2026.07c:temp0-1`` (one-shot offsets/tree/roles/ summaries/placement) to the D79 split: deterministic/anchor skeleton, sanity check, role pass, bottom-up summaries, and root-reduction placement. +The ``07f`` seed is input/seat identity only; provider output never mints. """ E0_SKELETON_VERSION: Final = ( @@ -682,6 +683,8 @@ def handle(self, *, work: ClaimedWork, meter: CostMeterPort) -> HandlerOutcome: self._persist_generation( source=source, sections=parsed, + blocks=blocks, + markdown=markdown, route=StructureRouteTag.PARSER_DEMOTED_CHECK, analysis=parsed_analysis, candidate_skeleton_hash=parsed_hash, @@ -714,6 +717,22 @@ def handle(self, *, work: ClaimedWork, meter: CostMeterPort) -> HandlerOutcome: else: route = StructureRouteTag.PARSER + if ( + current is not None + and current.route_tag == route + and current.skeleton_version == configured_skeleton_version + and current.skeleton_hash == skeleton_hash(sections=selected) + and current.roles_version == configured_role_version + and current.summary_version == self._summarizer.version + and current.placement_version == self._summarizer.placement_version + and current.placement_path is not None + and all(section.summary is not None for section in current.sections) + ): + # A checker-only bump already appended its check record. If it + # kept the route and selected tree, the complete current seats + # remain authoritative: no roles/summaries or pointer write. + return _chunk_outcome(work=work, source=source) + selected = self._assign_roles(sections=selected, meter=meter) if replay_summary is not None: @@ -727,6 +746,7 @@ def handle(self, *, work: ClaimedWork, meter: CostMeterPort) -> HandlerOutcome: sections=selected, placement_path=replay_placement, blocks=blocks, + markdown=markdown, ) else: summary_result = self._summarizer.summarize( @@ -739,6 +759,8 @@ def handle(self, *, work: ClaimedWork, meter: CostMeterPort) -> HandlerOutcome: self._persist_generation( source=source, sections=summary_result.sections, + blocks=blocks, + markdown=markdown, route=route, analysis=selection_analysis, candidate_skeleton_hash=selection_candidate_hash, @@ -752,23 +774,7 @@ def handle(self, *, work: ClaimedWork, meter: CostMeterPort) -> HandlerOutcome: producer_family=producer_family, make_current=True, ) - return HandlerOutcome( - follow_up=( - EnqueueWork( - deployment_id=work.deployment_id, - target_kind=work.target_kind, - target_id=work.target_id, - stage=PipelineStage.CHUNK, - component_version=E1_CHUNK_VERSION, - content_hash=work.content_hash, - lane=work.lane, - payload={ - "version_id": str(source.version_id), - "representation_id": str(source.representation_id), - }, - ), - ) - ) + return _chunk_outcome(work=work, source=source) def _fallback_with_terminal( self, @@ -1044,6 +1050,8 @@ def _persist_generation( *, source: StructureSource, sections: tuple[SnappedSection, ...], + blocks: tuple[Block, ...], + markdown: str, route: StructureRouteTag, analysis: SkeletonAnalysis, candidate_skeleton_hash: str, @@ -1059,29 +1067,20 @@ def _persist_generation( ) -> None: """Append one generation, then write its versioned sidecar. - The generation identity is seeded from the SELECTED tree and route — - never from the checker version. Both halves of the D79 rule follow - from one seed: a checker bump over an unchanged route+tree derives - the same id (ON CONFLICT no-op — no minted generation, no pointer - churn), while a genuine route flip or a different fallback tree - changes the seed and appends a real generation. Role, summary, and - placement versions/content stay in the seed because they change the - immutable section rows/sidecar. The checker version remains excluded. + The generation identity is selected-input and seat identity only, + never checker version or provider output. A checker bump over an + unchanged route+tree derives the same id, while a route/tree or seat + change appends a generation. Degraded markers still re-mint repairs. """ skeleton_version = _skeleton_version(settings=self._settings) - roles_hash = _roles_hash(sections=sections) - summaries_hash = _summaries_hash(sections=sections) - placement_hash = hashlib.sha256( - (placement_path or "").encode("utf-8") - ).hexdigest() generation_id = uuid5( NAMESPACE_URL, "rememberstack:structure:" f"{source.representation_id}:{E0_STRUCTURE_VERSION}:" f"{route.value}:{skeleton_hash(sections=sections)}:" - f"{skeleton_version}:{roles_version or 'none'}:{roles_hash}:" - f"{summary_version or 'degraded'}:{summaries_hash}:" - f"{placement_version or 'degraded'}:{placement_hash}", + f"{skeleton_version}:{roles_version or 'none'}:" + f"{summary_version or 'degraded'}:" + f"{placement_version or 'degraded'}", ) sidecar_key = ( source.blocks_uri.rsplit("/", 1)[0] @@ -1114,6 +1113,22 @@ def _persist_generation( make_current=make_current, ) ) + persisted_cache_keys = summary_cache_keys + if persisted.sections != sections or persisted.placement_path != placement_path: + if persisted.placement_path is not None and all( + section.summary is not None for section in persisted.sections + ): + persisted_cache_keys = self._summarizer.replay( + source=source, + sections=persisted.sections, + placement_path=persisted.placement_path, + blocks=blocks, + markdown=markdown, + ).cache_keys + else: + # A concurrent first write with different partial output did + # not necessarily see the prompts keyed by this attempt. + persisted_cache_keys = {} payload = _json_bytes( payload={ "structure_generation_id": str(persisted.structure_generation_id), @@ -1138,7 +1153,9 @@ def _persist_generation( "sections": [ { **section.model_dump(mode="json"), - "summary_cache_key": summary_cache_keys.get(section.node_path), + "summary_cache_key": persisted_cache_keys.get( + section.node_path + ), } for section in persisted.sections ], @@ -1152,6 +1169,27 @@ def _persist_generation( pass +def _chunk_outcome(*, work: ClaimedWork, source: StructureSource) -> HandlerOutcome: + """Continue the selected representation into deterministic chunking.""" + return HandlerOutcome( + follow_up=( + EnqueueWork( + deployment_id=work.deployment_id, + target_kind=work.target_kind, + target_id=work.target_id, + stage=PipelineStage.CHUNK, + component_version=E1_CHUNK_VERSION, + content_hash=work.content_hash, + lane=work.lane, + payload={ + "version_id": str(source.version_id), + "representation_id": str(source.representation_id), + }, + ), + ) + ) + + def _render_fallback_prompt( *, source: StructureSource, blocks: tuple[Block, ...], markdown: str, ceiling: int ) -> str: @@ -1379,22 +1417,6 @@ def _failure_envelope( } -def _roles_hash(*, sections: tuple[SnappedSection, ...]) -> str: - """Content identity for role-bearing immutable section rows.""" - payload = tuple((section.node_path, section.role) for section in sections) - return hashlib.sha256( - json.dumps(payload, separators=(",", ":")).encode("utf-8") - ).hexdigest() - - -def _summaries_hash(*, sections: tuple[SnappedSection, ...]) -> str: - """Content identity for ordered summary values, including nulls.""" - payload = tuple((section.node_path, section.summary) for section in sections) - return hashlib.sha256( - json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8") - ).hexdigest() - - def _is_incoherent(*, outcome: SkeletonCheckOutcome) -> bool: return outcome.value.startswith("incoherent_") diff --git a/src/rememberstack/workers/e0_summary.py b/src/rememberstack/workers/e0_summary.py index 0567a46b..ea7e530d 100644 --- a/src/rememberstack/workers/e0_summary.py +++ b/src/rememberstack/workers/e0_summary.py @@ -9,6 +9,7 @@ from __future__ import annotations from collections import defaultdict +from collections.abc import Callable from collections.abc import Iterable from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass @@ -23,6 +24,7 @@ from pydantic_settings import SettingsConfigDict from rememberstack.core import count_tokens +from rememberstack.core import LONG_TITLE from rememberstack.model import Block from rememberstack.model import ModelRequest from rememberstack.model import ObjectKey @@ -38,28 +40,33 @@ from rememberstack.spine.document_catalog import DocumentCatalog SUMMARY_CALL_TOKEN_CEILING: Final = 2_048 -"""Hard per-request ceiling under the pinned whitespace-token counter.""" +"""Per-request ceiling under the pinned whitespace-token counter.""" + +SUMMARY_CALL_CHAR_CEILING: Final = 16_384 +"""Hard per-request character ceiling for tokenizer-hostile source text.""" SUMMARY_BALANCED_FAN_IN: Final = 8 """Maximum ordered child/partial one-liners in one reduction call.""" SUMMARY_MAX_CHARS: Final = 512 -"""Closed response-schema ceiling for every one-line summary.""" +"""Post-parse ceiling for every normalized one-line provider value.""" SUMMARY_PARALLELISM: Final = 8 -"""Maximum independent same-depth section calls in flight.""" +"""Scheduling-only concurrency; unversioned because it cannot affect prompts.""" SUMMARY_TOKEN_COUNTER_VERSION: Final = "whitespace-token-counter-v1" """Names ``count_tokens`` semantics in summary-generation provenance.""" E0_SUMMARY_VERSION: Final = ( - "e0-summary-2026.07a:d79-bottom-up:block-shard-v1:" - f"{SUMMARY_TOKEN_COUNTER_VERSION}:ceiling{SUMMARY_CALL_TOKEN_CEILING}:" - f"balanced-fan-in{SUMMARY_BALANCED_FAN_IN}:line{SUMMARY_MAX_CHARS}" + "e0-summary-2026.07b:d79-bottom-up:block-shard-v2:" + f"{SUMMARY_TOKEN_COUNTER_VERSION}:token-ceiling{SUMMARY_CALL_TOKEN_CEILING}:" + f"char-ceiling{SUMMARY_CALL_CHAR_CEILING}:" + f"ceiling-reduction-v2:balanced-fan-in{SUMMARY_BALANCED_FAN_IN}:" + f"title-cap{LONG_TITLE}:normalized-line{SUMMARY_MAX_CHARS}:rendered-key-v1" ) """Bottom-up summarizer algorithm and all output-affecting constants.""" -E0_PLACEMENT_VERSION: Final = "e0-placement-2026.07a:d79-root-reduction-v1" +E0_PLACEMENT_VERSION: Final = "e0-placement-2026.07b:d79-root-reduction-v2" """D39 advisory placement emitted by the document-level summary reduction.""" _SUMMARY_INSTRUCTION: Final = """Write exactly one factual orientation line \ @@ -136,8 +143,11 @@ "temperature": 0.0, "token_counter": SUMMARY_TOKEN_COUNTER_VERSION, "token_ceiling": SUMMARY_CALL_TOKEN_CEILING, + "char_ceiling": SUMMARY_CALL_CHAR_CEILING, "balanced_fan_in": SUMMARY_BALANCED_FAN_IN, + "title_max_chars": LONG_TITLE, "summary_max_chars": SUMMARY_MAX_CHARS, + "placement_version": E0_PLACEMENT_VERSION, } @@ -257,12 +267,16 @@ def summarize( complete_child_lines = tuple( line for line in child_lines if line is not None ) + rendered_child_lines = _render_child_lines( + child_sections=child_sections, child_summaries=complete_child_lines + ) key = _summary_cache_key( section=section, direct_blocks=direct[section.node_path], - child_summaries=complete_child_lines, + child_lines=rendered_child_lines, model=self._settings.model, source_kind=source.source_kind, + markdown=markdown, ) cached = cache.get(key) if cached is not None: @@ -344,6 +358,7 @@ def replay( sections: tuple[SnappedSection, ...], placement_path: str, blocks: tuple[Block, ...], + markdown: str, ) -> SummaryResult: """Re-key stored successful summaries without calling the provider. @@ -372,14 +387,19 @@ def replay( ) if summary is None or any(value is None for value in child_summaries): raise ValueError("only complete summary generations can be replayed") - cache_keys[section.node_path] = _summary_cache_key( - section=section, - direct_blocks=direct[section.node_path], + rendered_child_lines = _render_child_lines( + child_sections=children.get(section.node_path, ()), child_summaries=tuple( value for value in child_summaries if value is not None ), + ) + cache_keys[section.node_path] = _summary_cache_key( + section=section, + direct_blocks=direct[section.node_path], + child_lines=rendered_child_lines, model=self._settings.model, source_kind=source.source_kind, + markdown=markdown, ) return SummaryResult( sections=sections, @@ -404,25 +424,15 @@ def _summarize_uncached( return _CallResult(summary=None, placement_path=None) events: list[_MeterEvent] = [] - child_lines = tuple( - f"{child.node_path} | {child.title} | {summary}" - for child, summary in zip(child_sections, child_summaries, strict=True) + child_lines = _render_child_lines( + child_sections=child_sections, child_summaries=child_summaries ) - reduced_children = self._reduce_lines( - section=section, - lines=child_lines, - call_kind="child-reduction", - call_key_prefix=f"summary:{section.node_path}:children", - events=events, - ) - if reduced_children is None: - return _CallResult(None, None, tuple(events)) final_prompt = _render_final_prompt( source=source, section=section, blocks=direct_blocks, - child_lines=reduced_children, + child_lines=child_lines, markdown=markdown, ) if _within_ceiling(prompt=final_prompt): @@ -451,13 +461,20 @@ def _summarize_uncached( if partial.summary is None: return _CallResult(None, None, tuple(events)) context_lines.append(f"direct-shard-{shard_index} | {partial.summary}") - context_lines.extend(reduced_children) + context_lines.extend(child_lines) reduced_context = self._reduce_lines( section=section, lines=tuple(context_lines), call_kind="context-reduction", call_key_prefix=f"summary:{section.node_path}:context", events=events, + render_final=lambda lines: _render_final_prompt( + source=source, + section=section, + blocks=(), + child_lines=lines, + markdown=markdown, + ), ) if reduced_context is None: return _CallResult(None, None, tuple(events)) @@ -480,20 +497,27 @@ def _reduce_lines( call_kind: str, call_key_prefix: str, events: list[_MeterEvent], + render_final: Callable[[tuple[str, ...]], str], ) -> tuple[str, ...] | None: - """Balanced fan-in reduction; every intermediate call is bounded.""" + """Reduce only until the rendered final prompt fits both ceilings.""" current = lines level = 0 - while len(current) > SUMMARY_BALANCED_FAN_IN: + used_singleton_pass = False + while not _within_ceiling(prompt=render_final(current)): + groups = _bounded_reduction_groups( + section=section, lines=current, call_kind=call_kind + ) + if groups is None: + return None + if all(len(group) == 1 for group in groups): + if used_singleton_pass: + return None + used_singleton_pass = True reduced: list[str] = [] - for group_index, group in enumerate( - _balanced_groups(values=current, fan_in=SUMMARY_BALANCED_FAN_IN) - ): + for group_index, group in enumerate(groups): prompt = _render_reduce_prompt( section=section, lines=group, call_kind=call_kind ) - if not _within_ceiling(prompt=prompt): - return None result = self._invoke_summary( prompt=prompt, call_key=(f"{call_key_prefix}:level:{level}:group:{group_index}"), @@ -529,10 +553,14 @@ def _invoke_final( usage=generated.usage, ) ) + normalized_summary = _normalize_one_line(generated.output.summary) + normalized_placement = _normalize_one_line( + generated.output.placement_path + ) + if normalized_summary is None or normalized_placement is None: + return _CallResult(None, None, tuple(events)) return _CallResult( - generated.output.summary, - generated.output.placement_path, - tuple(events), + normalized_summary, normalized_placement, tuple(events) ) generated_summary = self._model_provider.generate( request=ModelRequest( @@ -547,7 +575,10 @@ def _invoke_final( usage=generated_summary.usage, ) ) - return _CallResult(generated_summary.output.summary, None, tuple(events)) + normalized_summary = _normalize_one_line(generated_summary.output.summary) + if normalized_summary is None: + return _CallResult(None, None, tuple(events)) + return _CallResult(normalized_summary, None, tuple(events)) except ProviderCallError as error: if error.usage is not None: events.append( @@ -572,11 +603,11 @@ def _invoke_summary(self, *, prompt: str, call_key: str, tier: str) -> _CallResu ), response_type=SectionSummaryResponse, ) - return _CallResult( - generated.output.summary, - None, - (_MeterEvent(call_key=call_key, tier=tier, usage=generated.usage),), - ) + normalized_summary = _normalize_one_line(generated.output.summary) + event = _MeterEvent(call_key=call_key, tier=tier, usage=generated.usage) + if normalized_summary is None: + return _CallResult(None, None, (event,)) + return _CallResult(normalized_summary, None, (event,)) except ProviderCallError as error: if error.usage is not None: return _CallResult( @@ -597,36 +628,45 @@ def _invoke_summary(self, *, prompt: str, call_key: str, tier: str) -> _CallResu def _load_cache(self, *, doc_id: UUID) -> dict[str, _CacheValue]: """Read prior successful generation sidecars; corruption is a cache miss.""" cache: dict[str, _CacheValue] = {} - for uri in self._catalog.summary_cache_sidecars( - doc_id=doc_id, summary_version=self.version - ): + for uri in self._catalog.summary_cache_sidecars(doc_id=doc_id): try: payload = json.loads( self._artifact_store.read_bytes(key=ObjectKey(uri)) ) placement = payload.get("placement") - for section in payload.get("sections", ()): + sections = payload.get("sections", ()) + except Exception: # noqa: BLE001 - cache corruption is a miss + # Sidecars are immutable cache metadata, not an availability + # dependency. Missing/legacy/malformed objects simply miss. + continue + for section in sections: + try: key = section.get("summary_cache_key") summary = section.get("summary") if not isinstance(key, str) or not isinstance(summary, str): continue validated = SectionSummaryResponse(summary=summary) + normalized_summary = _normalize_one_line(validated.summary) + if normalized_summary is None: + continue cached_placement: str | None = None if section.get("node_path") == "0": validated_root = RootSummaryPlacementResponse( - summary=validated.summary, placement_path=placement + summary=normalized_summary, placement_path=placement ) - cached_placement = validated_root.placement_path + cached_placement = _normalize_one_line( + validated_root.placement_path + ) + if cached_placement is None: + continue cache.setdefault( key, _CacheValue( - summary=validated.summary, placement_path=cached_placement + summary=normalized_summary, placement_path=cached_placement ), ) - except Exception: # noqa: BLE001 - cache corruption is a miss - # Sidecars are immutable cache metadata, not an availability - # dependency. Missing/legacy/malformed objects simply miss. - continue + except Exception: # noqa: BLE001 - one bad entry is one miss + continue return cache @@ -666,22 +706,26 @@ def _summary_cache_key( *, section: SnappedSection, direct_blocks: tuple[Block, ...], - child_summaries: tuple[str, ...], + child_lines: tuple[str, ...], model: str, source_kind: str, + markdown: str, ) -> str: """The design's exact per-section content/configuration identity.""" child_hashes = tuple( - hashlib.sha256(summary.encode("utf-8")).hexdigest() - for summary in child_summaries + hashlib.sha256(line.encode("utf-8")).hexdigest() for line in child_lines ) prompt_hash = hashlib.sha256( json.dumps( { "contract": _SUMMARY_PROMPT_HASH, "node_path": section.node_path, - "title": section.title, - "source_kind": source_kind if section.node_path == "0" else None, + "title": _capped(section.title, fallback="(untitled section)"), + "source_kind": ( + _capped(source_kind, fallback="(unknown)") + if section.node_path == "0" + else None + ), }, ensure_ascii=False, separators=(",", ":"), @@ -689,10 +733,13 @@ def _summary_cache_key( ).encode("utf-8") ).hexdigest() payload = { - "ordered_constituent_block_hashes": tuple( - block.block_hash for block in direct_blocks + "ordered_rendered_block_hashes": tuple( + hashlib.sha256( + _render_block(block=block, markdown=markdown).encode("utf-8") + ).hexdigest() + for block in direct_blocks ), - "child_summary_hashes": child_hashes, + "child_line_hashes": child_hashes, "model": model, "prompt_hash": prompt_hash, "generation_params": _GENERATION_PARAMS, @@ -705,15 +752,16 @@ def _summary_cache_key( ).hexdigest() +def _render_block(*, block: Block, markdown: str) -> str: + return ( + f"[block ordinal={block.ordinal} type={block.type.value}" + f" hash={block.block_hash}]\n" + f"{markdown[block.char_start : block.char_end]}" + ) + + def _render_blocks(*, blocks: Iterable[Block], markdown: str) -> str: - rendered = [ - ( - f"[block ordinal={block.ordinal} type={block.type.value}" - f" hash={block.block_hash}]\n" - f"{markdown[block.char_start : block.char_end]}" - ) - for block in blocks - ] + rendered = [_render_block(block=block, markdown=markdown) for block in blocks] return "\n\n".join(rendered) if rendered else "(none)" @@ -728,15 +776,15 @@ def _render_final_prompt( if section.node_path == "0": return _ROOT_TEMPLATE.format( instruction=_ROOT_INSTRUCTION, - title=source.title or "(untitled)", - source_kind=source.source_kind, + title=_capped(source.title, fallback="(untitled)"), + source_kind=_capped(source.source_kind, fallback="(unknown)"), blocks=_render_blocks(blocks=blocks, markdown=markdown), children="\n".join(child_lines) if child_lines else "(none)", ) return _SECTION_TEMPLATE.format( instruction=_SUMMARY_INSTRUCTION, node_path=section.node_path, - title=section.title or "(untitled section)", + title=_capped(section.title, fallback="(untitled section)"), blocks=_render_blocks(blocks=blocks, markdown=markdown), children="\n".join(child_lines) if child_lines else "(none)", ) @@ -752,7 +800,7 @@ def _render_shard_prompt( return _SHARD_TEMPLATE.format( instruction=_SUMMARY_INSTRUCTION, node_path=section.node_path, - title=section.title or "(untitled section)", + title=_capped(section.title, fallback="(untitled section)"), shard_index=shard_index, blocks=_render_blocks(blocks=blocks, markdown=markdown), ) @@ -765,11 +813,25 @@ def _render_reduce_prompt( instruction=_REDUCE_INSTRUCTION, call_kind=call_kind, node_path=section.node_path, - title=section.title or "(untitled section)", + title=_capped(section.title, fallback="(untitled section)"), lines="\n".join(lines), ) +def _render_child_lines( + *, child_sections: tuple[SnappedSection, ...], child_summaries: tuple[str, ...] +) -> tuple[str, ...]: + return tuple( + f"{child.node_path} | " + f"{_capped(child.title, fallback='(untitled section)')} | {summary}" + for child, summary in zip(child_sections, child_summaries, strict=True) + ) + + +def _capped(value: str | None, *, fallback: str) -> str: + return (value or fallback)[:LONG_TITLE] + + def _block_shards( *, section: SnappedSection, blocks: tuple[Block, ...], markdown: str ) -> tuple[tuple[Block, ...], ...] | None: @@ -819,5 +881,43 @@ def _balanced_groups( return tuple(groups) +def _bounded_reduction_groups( + *, section: SnappedSection, lines: tuple[str, ...], call_kind: str +) -> tuple[tuple[str, ...], ...] | None: + """Choose the widest balanced groups whose rendered calls are bounded.""" + if not lines: + return None + for fan_in in range(min(SUMMARY_BALANCED_FAN_IN, len(lines)), 1, -1): + groups = _balanced_groups(values=lines, fan_in=fan_in) + if all( + _within_ceiling( + prompt=_render_reduce_prompt( + section=section, lines=group, call_kind=call_kind + ) + ) + for group in groups + ): + return groups + singleton_groups = tuple((line,) for line in lines) + if all( + _within_ceiling( + prompt=_render_reduce_prompt( + section=section, lines=group, call_kind=call_kind + ) + ) + for group in singleton_groups + ): + return singleton_groups + return None + + +def _normalize_one_line(value: str) -> str | None: + normalized = " ".join(value.split())[:SUMMARY_MAX_CHARS].rstrip() + return normalized or None + + def _within_ceiling(*, prompt: str) -> bool: - return count_tokens(text=prompt) <= SUMMARY_CALL_TOKEN_CEILING + return ( + count_tokens(text=prompt) <= SUMMARY_CALL_TOKEN_CEILING + and len(prompt) <= SUMMARY_CALL_CHAR_CEILING + ) diff --git a/src/tests/workers/test_d79_structure_route.py b/src/tests/workers/test_d79_structure_route.py index 4ecb6e2f..b1efadc3 100644 --- a/src/tests/workers/test_d79_structure_route.py +++ b/src/tests/workers/test_d79_structure_route.py @@ -76,11 +76,8 @@ def current_section_tree( assert representation_id == _REPRESENTATION return self.current - def summary_cache_sidecars( - self, *, doc_id: UUID, summary_version: str - ) -> tuple[str, ...]: + def summary_cache_sidecars(self, *, doc_id: UUID) -> tuple[str, ...]: assert doc_id == _DOC - assert summary_version return () def record_section_tree(self, *, record: SectionTreeRecord) -> PersistedSectionTree: @@ -429,9 +426,7 @@ def router(prompt: str, response_type: str) -> dict[str, object]: def test_checker_bump_does_not_mint_a_new_generation_when_route_holds( tmp_path: Path, ) -> None: - """The generation id is seeded from route + selected tree, never the - checker version (review: a checker bump must not churn generations or - re-chunk when the route does not flip).""" + """A checker-only bump appends its check and stops at the complete tree.""" source = "# A\n\nbody\n\n# B\n" store = LocalFSObjectStore(root=tmp_path) blocks = blockize(document_md=source) @@ -447,35 +442,56 @@ def test_checker_bump_does_not_mint_a_new_generation_when_route_holds( ).encode(), ) catalog = _Catalog() + summary_marker = {"value": "first"} provider = FakeModelProvider( generate_router=lambda prompt, response_type: ( {"verdict": "coherent"} if response_type == "SkeletonCheckResponse" else {"assignments": []} if response_type == "RoleClassificationResponse" - else {"summary": "Stable one-line summary."} + else {"summary": f"{summary_marker['value']} one-line summary."} if response_type == "SectionSummaryResponse" - else {"summary": "Stable root summary.", "placement_path": "/stable/topic/"} + else { + "summary": f"{summary_marker['value']} root summary.", + "placement_path": "/stable/topic/", + } ) ) - for checker_model in ("checker/generation-one", "checker/generation-two"): - StructureHandler( - catalog=catalog, # type: ignore[arg-type] - artifact_store=store, - model_provider=provider, - settings=StructurerSettings( - min_heading_density_per_10k=0, max_oversized_leaf_ratio=1 - ), - check_settings=SkeletonCheckSettings(model=checker_model), - role_settings=RoleSettings(model="roles/fake"), - ).handle(work=_work(), meter=NoopCostMeter()) - assert len(catalog.generations) == 2 # append-only recorder sees both runs + StructureHandler( + catalog=catalog, # type: ignore[arg-type] + artifact_store=store, + model_provider=provider, + settings=StructurerSettings( + min_heading_density_per_10k=0, max_oversized_leaf_ratio=1 + ), + check_settings=SkeletonCheckSettings(model="checker/generation-one"), + role_settings=RoleSettings(model="roles/fake"), + ).handle(work=_work(), meter=NoopCostMeter()) + first_generation_id = catalog.generations[0].structure_generation_id + summary_calls = sum( + "Call kind:" in request.prompt for request in provider.generated_requests + ) + summary_marker["value"] = "different" + StructureHandler( + catalog=catalog, # type: ignore[arg-type] + artifact_store=store, + model_provider=provider, + settings=StructurerSettings( + min_heading_density_per_10k=0, max_oversized_leaf_ratio=1 + ), + check_settings=SkeletonCheckSettings(model="checker/generation-two"), + role_settings=RoleSettings(model="roles/fake"), + ).handle(work=_work(), meter=NoopCostMeter()) + + assert len(catalog.generations) == 1 + assert catalog.generations[0].structure_generation_id == first_generation_id + assert len(catalog.checks) == 2 assert ( - len({record.structure_generation_id for record in catalog.generations}) == 1 - ) # ...but they derive the SAME id: ON CONFLICT makes the second a no-op - assert len(catalog.checks) == 2 # while every check outcome is ledgered - assert all(record.summary_version is not None for record in catalog.generations) - assert all(record.placement_version is not None for record in catalog.generations) + sum("Call kind:" in request.prompt for request in provider.generated_requests) + == summary_calls + ) + assert catalog.generations[0].summary_version is not None + assert catalog.generations[0].placement_version is not None def test_role_classifier_failure_defaults_to_body(tmp_path: Path) -> None: diff --git a/src/tests/workers/test_d79_summaries.py b/src/tests/workers/test_d79_summaries.py index 4314b5ef..cfd00a1d 100644 --- a/src/tests/workers/test_d79_summaries.py +++ b/src/tests/workers/test_d79_summaries.py @@ -9,12 +9,17 @@ from rememberstack.core import blockize from rememberstack.core import count_tokens from rememberstack.core import parse_heading_skeleton +from rememberstack.model import ObjectKey from rememberstack.model import RootSummaryPlacementResponse from rememberstack.model import SectionSummaryResponse from rememberstack.model import StructureSource +from rememberstack.workers.e0_summary import _render_child_lines +from rememberstack.workers.e0_summary import _summary_cache_key from rememberstack.workers.e0_summary import SectionSummarizer from rememberstack.workers.e0_summary import SUMMARY_BALANCED_FAN_IN +from rememberstack.workers.e0_summary import SUMMARY_CALL_CHAR_CEILING from rememberstack.workers.e0_summary import SUMMARY_CALL_TOKEN_CEILING +from rememberstack.workers.e0_summary import SUMMARY_MAX_CHARS from rememberstack.workers.e0_summary import SummarySettings _DEPLOYMENT = UUID("71000000-0000-0000-0000-000000000001") @@ -26,11 +31,8 @@ class _NoCacheCatalog: """The unit proofs exercise composition without prior sidecars.""" - def summary_cache_sidecars( - self, *, doc_id: UUID, summary_version: str - ) -> tuple[str, ...]: + def summary_cache_sidecars(self, *, doc_id: UUID) -> tuple[str, ...]: assert doc_id == _DOC - assert summary_version return () @@ -168,7 +170,6 @@ def test_many_children_use_balanced_bounded_fan_in(tmp_path: Path) -> None: markdown = "\n\n".join( ( "# Parent", - "Parent preamble.", *( value for index in range(18) @@ -180,11 +181,11 @@ def test_many_children_use_balanced_bounded_fan_in(tmp_path: Path) -> None: def route(prompt: str, response_type: str) -> dict[str, object]: if response_type == "RootSummaryPlacementResponse": return {"summary": "Wide document.", "placement_path": "/proof/wide/"} - if "Call kind: child-reduction" in prompt: + if "Call kind: context-reduction" in prompt: return {"summary": "Balanced child group."} if "Section path: 0.0\n" in prompt: return {"summary": "Wide parent."} - return {"summary": "Child line."} + return {"summary": "x " * 400} result, requests = _summarize( tmp_path=tmp_path, @@ -194,7 +195,7 @@ def route(prompt: str, response_type: str) -> dict[str, object]: reductions = [ request.prompt for request in requests - if "Call kind: child-reduction" in request.prompt + if "Call kind: context-reduction" in request.prompt and "Section path: 0.0\n" in request.prompt ] assert len(reductions) == 3 @@ -210,11 +211,243 @@ def route(prompt: str, response_type: str) -> dict[str, object]: assert result.sections[1].summary == "Wide parent." -def test_summary_response_schemas_are_closed_and_bounded() -> None: +def test_eight_fat_child_lines_reduce_below_fan_in_and_complete(tmp_path: Path) -> None: + markdown = "\n\n".join( + ( + "# Parent", + *( + value + for index in range(8) + for value in (f"## Child {index}", f"Body {index}.") + ), + ) + ) + + def route(prompt: str, response_type: str) -> dict[str, object]: + if response_type == "RootSummaryPlacementResponse": + return {"summary": "Fat document.", "placement_path": "/proof/fat/"} + if "Call kind: context-reduction" in prompt: + return {"summary": "Compressed child group."} + if "Section path: 0.0\n" in prompt: + return {"summary": "Recovered fat parent."} + return {"summary": "x " * 400} + + result, requests = _summarize( + tmp_path=tmp_path, + markdown=markdown, + provider=FakeModelProvider(generate_router=route), + ) + + reductions = [ + request.prompt + for request in requests + if "Call kind: context-reduction" in request.prompt + and "Section path: 0.0\n" in request.prompt + ] + assert reductions + assert all( + len(prompt.split("Ordered one-liners:\n", 1)[1].splitlines()) + < SUMMARY_BALANCED_FAN_IN + for prompt in reductions + ) + assert result.sections[1].summary == "Recovered fat parent." + assert result.summary_version is not None + + +def test_nine_short_children_compose_in_one_call_without_reduction( + tmp_path: Path, +) -> None: + markdown = "\n\n".join( + ( + "# Parent", + *( + value + for index in range(9) + for value in (f"## Child {index}", f"Body {index}.") + ), + ) + ) + + def route(prompt: str, response_type: str) -> dict[str, object]: + if response_type == "RootSummaryPlacementResponse": + return {"summary": "Short document.", "placement_path": "/proof/short/"} + if "Section path: 0.0\n" in prompt: + assert prompt.count(" | Child ") == 9 + return {"summary": "Nine children in one pass."} + return {"summary": "Short child line."} + + result, requests = _summarize( + tmp_path=tmp_path, + markdown=markdown, + provider=FakeModelProvider(generate_router=route), + ) + + parent_requests = [ + request.prompt + for request in requests + if "Section path: 0.0\n" in request.prompt + ] + assert len(parent_requests) == 1 + assert "Call kind: section-final" in parent_requests[0] + assert not any("reduction" in request.prompt for request in requests) + assert result.sections[1].summary == "Nine children in one pass." + + +def test_three_thousand_word_heading_is_capped_and_document_completes( + tmp_path: Path, +) -> None: + markdown = f"# {'heading ' * 3_000}\n\nBody remains small." + provider = FakeModelProvider( + generate_payloads={ + "SectionSummaryResponse": {"summary": "Bounded headed section."}, + "RootSummaryPlacementResponse": { + "summary": "Bounded headed document.", + "placement_path": "/proof/heading/", + }, + } + ) + + result, requests = _summarize( + tmp_path=tmp_path, markdown=markdown, provider=provider + ) + + assert result.summary_version is not None + assert all(section.summary is not None for section in result.sections) + assert requests + assert all(len(request.prompt) <= SUMMARY_CALL_CHAR_CEILING for request in requests) + + +def test_whitespace_poor_text_never_crosses_the_character_ceiling( + tmp_path: Path, +) -> None: + markdown = "\n\n".join(("# Dense", *("界" * 2_000 for _ in range(20)))) + + def route(prompt: str, response_type: str) -> dict[str, object]: + if response_type == "RootSummaryPlacementResponse": + return {"summary": "Dense document.", "placement_path": "/proof/dense/"} + if "Call kind: block-shard" in prompt: + return {"summary": "Dense bounded shard."} + return {"summary": "Dense section."} + + result, requests = _summarize( + tmp_path=tmp_path, + markdown=markdown, + provider=FakeModelProvider(generate_router=route), + ) + + assert ( + len( + [ + request + for request in requests + if "Call kind: block-shard" in request.prompt + ] + ) + > 1 + ) + assert all(len(request.prompt) <= SUMMARY_CALL_CHAR_CEILING for request in requests) + assert result.summary_version is not None + + +def test_overlong_multiline_provider_values_are_normalized_after_parse( + tmp_path: Path, +) -> None: + raw = " first line \n second line " + ("tail " * 200) + provider = FakeModelProvider( + generate_payloads={ + "SectionSummaryResponse": {"summary": raw}, + "RootSummaryPlacementResponse": { + "summary": raw, + "placement_path": "/proof/\n" + ("long-segment " * 100), + }, + } + ) + + result, _ = _summarize( + tmp_path=tmp_path, markdown="# Normalized\n\nBody.", provider=provider + ) + + values = [*(section.summary for section in result.sections), result.placement_path] + assert result.summary_version is not None + assert all(value is not None for value in values) + assert all("\n" not in value for value in values if value is not None) + assert all(len(value) <= SUMMARY_MAX_CHARS for value in values if value is not None) + + +def test_parent_cache_key_hashes_rendered_child_titles() -> None: + markdown = "# Parent\n\n## Old child\n\nBody." + blocks = blockize(document_md=markdown) + sections = parse_heading_skeleton( + blocks=blocks, title="Cache key proof", markdown_chars=len(markdown) + ) + parent = sections[1] + child = sections[2] + old_lines = _render_child_lines( + child_sections=(child,), child_summaries=("Unchanged summary.",) + ) + renamed_lines = _render_child_lines( + child_sections=(child.model_copy(update={"title": "Renamed child"}),), + child_summaries=("Unchanged summary.",), + ) + + old_key = _summary_cache_key( + section=parent, + direct_blocks=(), + child_lines=old_lines, + model="summary/test", + source_kind="upload", + markdown=markdown, + ) + renamed_key = _summary_cache_key( + section=parent, + direct_blocks=(), + child_lines=renamed_lines, + model="summary/test", + source_kind="upload", + markdown=markdown, + ) + assert old_key != renamed_key + + +def test_one_bad_sidecar_entry_does_not_discard_later_cache_values( + tmp_path: Path, +) -> None: + class _SidecarCatalog: + def summary_cache_sidecars(self, *, doc_id: UUID) -> tuple[str, ...]: + assert doc_id == _DOC + return ("cache/pageindex.json",) + + store = LocalFSObjectStore(root=tmp_path) + store.write_bytes( + key=ObjectKey("cache/pageindex.json"), + content=( + b'{"placement":null,"sections":[' + b'{"node_path":"0","summary_cache_key":"bad","summary":"root"},' + b'{"node_path":"0.0","summary_cache_key":"good","summary":"usable"}]}' + ), + ) + summarizer = SectionSummarizer( + catalog=_SidecarCatalog(), # type: ignore[arg-type] + artifact_store=store, + model_provider=None, + settings=SummarySettings(model="summary/test"), + ) + + cache = summarizer._load_cache(doc_id=_DOC) + assert set(cache) == {"good"} + assert cache["good"].summary == "usable" + + +def test_summary_response_schemas_are_closed_and_provider_compatible() -> None: section_schema = SectionSummaryResponse.model_json_schema() root_schema = RootSummaryPlacementResponse.model_json_schema() assert section_schema["additionalProperties"] is False assert root_schema["additionalProperties"] is False - assert section_schema["properties"]["summary"]["maxLength"] == 512 - assert root_schema["properties"]["summary"]["maxLength"] == 512 - assert root_schema["properties"]["placement_path"]["maxLength"] == 512 + for field in ( + section_schema["properties"]["summary"], + root_schema["properties"]["summary"], + root_schema["properties"]["placement_path"], + ): + assert field["minLength"] == 1 + assert "maxLength" not in field + assert "pattern" not in field diff --git a/src/tests/workers/test_e0_chain.py b/src/tests/workers/test_e0_chain.py index 5766b8fc..782a446e 100644 --- a/src/tests/workers/test_e0_chain.py +++ b/src/tests/workers/test_e0_chain.py @@ -33,6 +33,7 @@ from rememberstack.model import PipelineStage from rememberstack.model import ProcessingLane from rememberstack.model import ProcessingTarget +from rememberstack.model import ProviderCallError from rememberstack.model import RunResultOutcome from rememberstack.model import SectionTreeRecord from rememberstack.model import SkeletonStats @@ -505,6 +506,26 @@ def _structure_worker( return Worker(ledger=rig.ledger, registry=registry) +def _structure_work( + *, version_id: UUID, representation_id: UUID, content_hash: str +) -> ClaimedWork: + return ClaimedWork( + processing_id=uuid4(), + deployment_id=_DEPLOYMENT_ID, + target_kind=ProcessingTarget.DOCUMENT_VERSION, + target_id=version_id, + stage=PipelineStage.STRUCTURE, + component_version=E0_STRUCTURE_VERSION, + content_hash=content_hash, + lane=ProcessingLane.STEADY, + attempt=1, + payload={ + "version_id": str(version_id), + "representation_id": str(representation_id), + }, + ) + + def test_full_structure_route_persists_summaries_and_root_placement( rig: _E0Rig, ) -> None: @@ -773,6 +794,347 @@ def route(prompt: str, response_type: str) -> dict[str, object]: assert rows == {"generations": 2, "complete": 2} +def test_complete_runs_with_different_outputs_keep_first_generation_truth( + rig: _E0Rig, monkeypatch: pytest.MonkeyPatch +) -> None: + """Input/seat identity collides even when two full providers disagree.""" + + def provider(marker: str) -> FakeModelProvider: + def route(prompt: str, response_type: str) -> dict[str, object]: + if response_type == "SkeletonCheckResponse": + return {"verdict": "coherent"} + if response_type == "RoleClassificationResponse": + return {"assignments": []} + if response_type == "RootSummaryPlacementResponse": + return { + "summary": f"{marker} root summary.", + "placement_path": f"/field-research/{marker}/", + } + path = prompt.split("Section path: ", 1)[1].splitlines()[0] + return {"summary": f"{marker} summary for {path}."} + + return FakeModelProvider(generate_router=route) + + ingested = rig.ingestor.ingest( + deployment_id=_DEPLOYMENT_ID, + upload=DocumentUpload( + filename="identity.md", + mime="text/markdown", + content=_STRUCTURED_SOURCE.encode("utf-8"), + ), + ) + assert rig.run(stage=PipelineStage.CONVERT) is RunResultOutcome.SUCCEEDED + representation = rig.row( + sql="SELECT representation_id FROM document_representations" + " WHERE version_id = :version_id", + params={"version_id": ingested.version_id}, + ) + representation_id = representation["representation_id"] + work = _structure_work( + version_id=ingested.version_id, + representation_id=representation_id, # type: ignore[arg-type] + content_hash=ingested.content_hash, + ) + first_handler = StructureHandler( + catalog=rig.catalog, + artifact_store=rig.artifact_store, + model_provider=provider("first"), + settings=StructurerSettings(min_blocks_for_llm=3), + summary_settings=SummarySettings(model="summary/identity-proof"), + ) + first_handler.handle(work=work, meter=NoopCostMeter()) + first_current = rig.row( + sql="SELECT current_structure_generation_id, pageindex_uri" + " FROM document_representations" + " WHERE representation_id = :representation_id", + params={"representation_id": representation_id}, + ) + first_generation_id = first_current["current_structure_generation_id"] + + # Simulate two attempts that both observed no generation, with the first + # attempt's PostgreSQL commit surviving but its sidecar write missing. + rig.artifact_store.purge_objects( + keys=(ObjectKey(str(first_current["pageindex_uri"])),), prefixes=() + ) + monkeypatch.setattr( + rig.catalog, "current_section_tree", lambda *, representation_id: None + ) + monkeypatch.setattr(rig.catalog, "summary_cache_sidecars", lambda *, doc_id: ()) + second_handler = StructureHandler( + catalog=rig.catalog, + artifact_store=rig.artifact_store, + model_provider=provider("different"), + settings=StructurerSettings(min_blocks_for_llm=3), + summary_settings=SummarySettings(model="summary/identity-proof"), + ) + second_handler.handle(work=work, meter=NoopCostMeter()) + + rows = rig.row( + sql="SELECT count(*) AS generations," + " min(structure_generation_id::text) AS generation_id" + " FROM document_structure_generations" + " WHERE representation_id = :representation_id", + params={"representation_id": representation_id}, + ) + assert rows == {"generations": 1, "generation_id": str(first_generation_id)} + with rig.engine.connect() as connection: + summaries = connection.execute( + text( + "SELECT summary FROM document_sections" + " WHERE structure_generation_id = :generation_id" + " ORDER BY ordinal" + ), + {"generation_id": first_generation_id}, + ).scalars() + assert all(str(summary).startswith("first ") for summary in summaries) + repaired_sidecar = json.loads( + rig.artifact_store.read_bytes( + key=ObjectKey(str(first_current["pageindex_uri"])) + ) + ) + assert repaired_sidecar["structure_generation_id"] == str(first_generation_id) + assert all( + str(section["summary"]).startswith("first ") + for section in repaired_sidecar["sections"] + ) + + +def test_identical_handler_replay_rewrites_missing_sidecar_without_calls( + rig: _E0Rig, +) -> None: + """A same-seat retry replays the first generation's stored output.""" + provider = FakeModelProvider( + generate_payloads={ + "SkeletonCheckResponse": {"verdict": "coherent"}, + "RoleClassificationResponse": {"assignments": []}, + "SectionSummaryResponse": {"summary": "Replay section summary."}, + "RootSummaryPlacementResponse": { + "summary": "Replay root summary.", + "placement_path": "/field-research/replay/", + }, + } + ) + ingested = rig.ingestor.ingest( + deployment_id=_DEPLOYMENT_ID, + upload=DocumentUpload( + filename="replay.md", + mime="text/markdown", + content=_STRUCTURED_SOURCE.encode("utf-8"), + ), + ) + assert rig.run(stage=PipelineStage.CONVERT) is RunResultOutcome.SUCCEEDED + representation = rig.row( + sql="SELECT representation_id FROM document_representations" + " WHERE version_id = :version_id", + params={"version_id": ingested.version_id}, + ) + representation_id = representation["representation_id"] + handler = StructureHandler( + catalog=rig.catalog, + artifact_store=rig.artifact_store, + model_provider=provider, + settings=StructurerSettings(min_blocks_for_llm=3), + summary_settings=SummarySettings(model="summary/replay-proof"), + ) + work = _structure_work( + version_id=ingested.version_id, + representation_id=representation_id, # type: ignore[arg-type] + content_hash=ingested.content_hash, + ) + handler.handle(work=work, meter=NoopCostMeter()) + current = rig.row( + sql="SELECT current_structure_generation_id, pageindex_uri" + " FROM document_representations" + " WHERE representation_id = :representation_id", + params={"representation_id": representation_id}, + ) + first_generation_id = current["current_structure_generation_id"] + first_call_count = len(provider.generated_requests) + sidecar_key = ObjectKey(str(current["pageindex_uri"])) + rig.artifact_store.purge_objects(keys=(sidecar_key,), prefixes=()) + + handler.handle(work=work, meter=NoopCostMeter()) + + assert len(provider.generated_requests) == first_call_count + after = rig.row( + sql="SELECT current_structure_generation_id," + " (SELECT count(*) FROM document_structure_generations" + " WHERE representation_id = :representation_id) AS generations" + " FROM document_representations" + " WHERE representation_id = :representation_id", + params={"representation_id": representation_id}, + ) + assert after == { + "current_structure_generation_id": first_generation_id, + "generations": 1, + } + replayed_sidecar = json.loads(rig.artifact_store.read_bytes(key=sidecar_key)) + assert replayed_sidecar["structure_generation_id"] == str(first_generation_id) + assert all(section["summary_cache_key"] for section in replayed_sidecar["sections"]) + + +def test_degraded_sidecar_repairs_only_failed_branch_and_ancestors(rig: _E0Rig) -> None: + """A failed mid-tree node preserves and later reuses its independent peers.""" + source = "\n\n".join( + ( + "# Broken branch", + "Broken branch preamble.", + "## Successful child", + "Successful child body.", + "# Healthy sibling", + "Healthy sibling body.", + ) + ) + first_calls: list[str] = [] + + def failing_route(prompt: str, response_type: str) -> dict[str, object]: + if response_type == "SkeletonCheckResponse": + return {"verdict": "coherent"} + if response_type == "RoleClassificationResponse": + return {"assignments": []} + path = ( + "0" + if response_type == "RootSummaryPlacementResponse" + else prompt.split("Section path: ", 1)[1].splitlines()[0] + ) + first_calls.append(path) + if path == "0.0": + raise ProviderCallError("only the branch summary fails") + if response_type == "RootSummaryPlacementResponse": + return {"summary": "Unexpected root.", "placement_path": "/unexpected/"} + return {"summary": f"Stable summary for {path}."} + + ingested = rig.ingestor.ingest( + deployment_id=_DEPLOYMENT_ID, + upload=DocumentUpload( + filename="partial.md", mime="text/markdown", content=source.encode("utf-8") + ), + ) + assert rig.run(stage=PipelineStage.CONVERT) is RunResultOutcome.SUCCEEDED + first_worker = _structure_worker( + rig, + FakeModelProvider(generate_router=failing_route), + summary_model="summary/repair-proof", + ) + assert ( + first_worker.run_one( + deployment_id=_DEPLOYMENT_ID, + stage=PipelineStage.STRUCTURE, + lane=ProcessingLane.STEADY, + ).outcome + is RunResultOutcome.SUCCEEDED + ) + representation = rig.row( + sql="SELECT representation_id, current_structure_generation_id," + " pageindex_uri FROM document_representations" + " WHERE version_id = :version_id", + params={"version_id": ingested.version_id}, + ) + degraded_generation_id = representation["current_structure_generation_id"] + with rig.engine.connect() as connection: + degraded_rows = ( + connection.execute( + text( + "SELECT node_path, summary, placement_path" + " FROM document_sections" + " WHERE structure_generation_id = :generation_id" + " ORDER BY ordinal" + ), + {"generation_id": degraded_generation_id}, + ) + .mappings() + .all() + ) + degraded_slots = ( + connection.execute( + text( + "SELECT summary_version, placement_version" + " FROM document_structure_generations" + " WHERE structure_generation_id = :generation_id" + ), + {"generation_id": degraded_generation_id}, + ) + .mappings() + .one() + ) + by_path = {row["node_path"]: row for row in degraded_rows} + assert first_calls[0] == "0.0.0" + assert set(first_calls[1:]) == {"0.0", "0.1"} + assert by_path["0.0.0"]["summary"] == "Stable summary for 0.0.0." + assert by_path["0.1"]["summary"] == "Stable summary for 0.1." + assert by_path["0.0"]["summary"] is None + assert by_path["0"]["summary"] is None + assert all(row["placement_path"] is None for row in degraded_rows) + assert dict(degraded_slots) == {"summary_version": None, "placement_version": None} + degraded_sidecar = json.loads( + rig.artifact_store.read_bytes( + key=ObjectKey(str(representation["pageindex_uri"])) + ) + ) + degraded_sidecar_by_path = { + section["node_path"]: section for section in degraded_sidecar["sections"] + } + assert degraded_sidecar["generations"]["summary"] is None + assert degraded_sidecar["generations"]["placement"] is None + assert degraded_sidecar_by_path["0.0"]["summary"] is None + assert degraded_sidecar_by_path["0"]["summary"] is None + assert degraded_sidecar_by_path["0.0.0"]["summary_cache_key"] + assert degraded_sidecar_by_path["0.1"]["summary_cache_key"] + + repair_calls: list[str] = [] + + def healthy_route(prompt: str, response_type: str) -> dict[str, object]: + path = ( + "0" + if response_type == "RootSummaryPlacementResponse" + else prompt.split("Section path: ", 1)[1].splitlines()[0] + ) + repair_calls.append(path) + if response_type == "RootSummaryPlacementResponse": + return { + "summary": "Repaired root summary.", + "placement_path": "/field-research/repaired/", + } + return {"summary": f"Repaired summary for {path}."} + + StructureHandler( + catalog=rig.catalog, + artifact_store=rig.artifact_store, + model_provider=FakeModelProvider(generate_router=healthy_route), + settings=StructurerSettings(min_blocks_for_llm=3), + summary_settings=SummarySettings(model="summary/repair-proof"), + ).handle( + work=_structure_work( + version_id=ingested.version_id, + representation_id=representation["representation_id"], # type: ignore[arg-type] + content_hash=ingested.content_hash, + ), + meter=NoopCostMeter(), + ) + + assert repair_calls == ["0.0", "0"] + repaired = rig.row( + sql="SELECT r.current_structure_generation_id, g.summary_version," + " g.placement_version, s.summary, s.placement_path," + " (SELECT count(*) FROM document_structure_generations" + " WHERE representation_id = r.representation_id) AS generations" + " FROM document_representations r" + " JOIN document_structure_generations g" + " ON g.structure_generation_id = r.current_structure_generation_id" + " JOIN document_sections s" + " ON s.structure_generation_id = g.structure_generation_id" + " AND s.node_path = '0'" + " WHERE r.representation_id = :representation_id", + params={"representation_id": representation["representation_id"]}, + ) + assert repaired["current_structure_generation_id"] != degraded_generation_id + assert repaired["summary_version"] is not None + assert repaired["placement_version"] is not None + assert repaired["summary"] == "Repaired root summary." + assert repaired["placement_path"] == "/field-research/repaired/" + assert repaired["generations"] == 2 + + def test_summary_seat_swap_copies_skeleton_and_moves_pointer( rig: _E0Rig, monkeypatch: pytest.MonkeyPatch ) -> None: diff --git a/website/src/app/docs/deployment/page.mdx b/website/src/app/docs/deployment/page.mdx index 22765aaa..0ce30575 100644 --- a/website/src/app/docs/deployment/page.mdx +++ b/website/src/app/docs/deployment/page.mdx @@ -25,7 +25,9 @@ placement reduction, default `z-ai/glm-4.7-flash`), and `REMEMBERSTACK_STRUCTURER_MODEL` — which names **only** the string-anchor fallback proposer for documents whose headings cannot be parsed or whose parsed tree fails the sanity check. Routine skeleton construction is -deterministic; bounded summary calls still run on the selected tree. +deterministic; bounded summary calls still run on the selected tree. Their +versioned token estimate is whitespace-based, while a companion character +ceiling is the hard request bound for whitespace-poor input. ## Start the stack