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..9a837f46 100644 --- a/plan/designs/e0_files_design.md +++ b/plan/designs/e0_files_design.md @@ -382,10 +382,20 @@ 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. + 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 + 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..aba97cd1 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 section summary and no auxiliary prose.""" + + model_config = ConfigDict(frozen=True, extra="forbid") + + summary: str = Field(min_length=1) + + +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) + placement_path: str = Field(min_length=1) + + 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..053a85cd 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,41 @@ 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) -> 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} + ).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 +483,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 +731,31 @@ 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 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 +767,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..e6b01b73 100644 --- a/src/rememberstack/workers/e0.py +++ b/src/rememberstack/workers/e0.py @@ -78,17 +78,20 @@ 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.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, and role pass. Summary and placement generation slots remain empty. +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 = ( @@ -538,14 +541,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 +569,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,41 +665,116 @@ 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 - - with_roles = self._assign_roles(sections=selected, meter=meter) + 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, + blocks=blocks, + markdown=markdown, + 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 + + 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: + 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, + markdown=markdown, + ) + 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, + blocks=blocks, + markdown=markdown, 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, ) - 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, @@ -967,24 +1050,27 @@ def _persist_generation( *, source: StructureSource, sections: tuple[SnappedSection, ...], + blocks: tuple[Block, ...], + markdown: str, route: StructureRouteTag, analysis: SkeletonAnalysis, candidate_skeleton_hash: str, 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: """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 version stays - in the seed because roles change the section rows themselves. + 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) generation_id = uuid5( @@ -992,7 +1078,9 @@ def _persist_generation( "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'}:" + f"{summary_version or 'degraded'}:" + f"{placement_version or 'degraded'}", ) sidecar_key = ( source.blocks_uri.rsplit("/", 1)[0] @@ -1006,7 +1094,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 +1102,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, @@ -1023,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), @@ -1045,7 +1151,13 @@ 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": persisted_cache_keys.get( + section.node_path + ), + } + for section in persisted.sections ], } ) @@ -1057,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: diff --git a/src/rememberstack/workers/e0_summary.py b/src/rememberstack/workers/e0_summary.py new file mode 100644 index 00000000..ea7e530d --- /dev/null +++ b/src/rememberstack/workers/e0_summary.py @@ -0,0 +1,923 @@ +"""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 Callable +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.core import LONG_TITLE +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 +"""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 +"""Post-parse ceiling for every normalized one-line provider value.""" + +SUMMARY_PARALLELISM: Final = 8 +"""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.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.07b:d79-root-reduction-v2" +"""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, + "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, +} + + +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 + ) + 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_lines=rendered_child_lines, + model=self._settings.model, + source_kind=source.source_kind, + markdown=markdown, + ) + 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, ...], + markdown: str, + ) -> 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") + 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, + 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 = _render_child_lines( + child_sections=child_sections, child_summaries=child_summaries + ) + + final_prompt = _render_final_prompt( + source=source, + section=section, + blocks=direct_blocks, + child_lines=child_lines, + 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(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)) + 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], + render_final: Callable[[tuple[str, ...]], str], + ) -> tuple[str, ...] | None: + """Reduce only until the rendered final prompt fits both ceilings.""" + current = lines + level = 0 + 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(groups): + prompt = _render_reduce_prompt( + section=section, lines=group, call_kind=call_kind + ) + 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, + ) + ) + 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( + normalized_summary, normalized_placement, 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, + ) + ) + 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( + _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, + ) + 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( + 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): + try: + payload = json.loads( + self._artifact_store.read_bytes(key=ObjectKey(uri)) + ) + placement = payload.get("placement") + 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=normalized_summary, placement_path=placement + ) + cached_placement = _normalize_one_line( + validated_root.placement_path + ) + if cached_placement is None: + continue + cache.setdefault( + key, + _CacheValue( + summary=normalized_summary, placement_path=cached_placement + ), + ) + except Exception: # noqa: BLE001 - one bad entry is one 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_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(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": _capped(section.title, fallback="(untitled section)"), + "source_kind": ( + _capped(source_kind, fallback="(unknown)") + if section.node_path == "0" + else None + ), + }, + ensure_ascii=False, + separators=(",", ":"), + sort_keys=True, + ).encode("utf-8") + ).hexdigest() + payload = { + "ordered_rendered_block_hashes": tuple( + hashlib.sha256( + _render_block(block=block, markdown=markdown).encode("utf-8") + ).hexdigest() + for block in direct_blocks + ), + "child_line_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_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 = [_render_block(block=block, markdown=markdown) 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=_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=_capped(section.title, fallback="(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=_capped(section.title, fallback="(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=_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: + """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 _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 + and len(prompt) <= SUMMARY_CALL_CHAR_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..b1efadc3 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,19 @@ 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) -> tuple[str, ...]: + assert doc_id == _DOC + 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 +101,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 +139,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 @@ -405,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) @@ -423,22 +442,56 @@ def test_checker_bump_does_not_mint_a_new_generation_when_route_holds( ).encode(), ) catalog = _Catalog() - for checker_model in ("checker/generation-one", "checker/generation-two"): - StructureHandler( - catalog=catalog, # type: ignore[arg-type] - artifact_store=store, - model_provider=None, - 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 + 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": f"{summary_marker['value']} one-line summary."} + if response_type == "SectionSummaryResponse" + else { + "summary": f"{summary_marker['value']} root summary.", + "placement_path": "/stable/topic/", + } + ) + ) + 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 + 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 new file mode 100644 index 00000000..cfd00a1d --- /dev/null +++ b/src/tests/workers/test_d79_summaries.py @@ -0,0 +1,453 @@ +"""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 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") +_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) -> tuple[str, ...]: + assert doc_id == _DOC + 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", + *( + 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: context-reduction" in prompt: + return {"summary": "Balanced child group."} + if "Section path: 0.0\n" in prompt: + return {"summary": "Wide 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 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_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 + 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 e285be13..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 @@ -41,14 +42,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 +488,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 +500,45 @@ 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 _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: + """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 +563,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 +582,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 +602,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 +667,648 @@ 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_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: + """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..0ce30575 100644 --- a/website/src/app/docs/deployment/page.mdx +++ b/website/src/app/docs/deployment/page.mdx @@ -17,13 +17,17 @@ 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. Their +versioned token estimate is whitespace-based, while a companion character +ceiling is the hard request bound for whitespace-poor input. ## Start the stack