diff --git a/DOCS-MAP.md b/DOCS-MAP.md index 7f87d825..8716f6d7 100644 --- a/DOCS-MAP.md +++ b/DOCS-MAP.md @@ -1,9 +1,9 @@ - + # DOCS-MAP - memorymaster # Covers: trust verdicts and replacements for every canonical documentation surface. # Key terms: CURRENT, SUPERSEDED, workflow intelligence, compiled user profile, roadmap, ADR. # Read when: locating authoritative project documentation before reading doc bodies. -# Updated: 2026-09-28 for shared CLI navigation; earlier dated entries retain their recorded scope. +# Updated: 2026-10-22 for AO-3 source configuration; earlier entries retain their dated scope. # Rule: generated profiles are disposable projections; MemoryMaster remains authority. @@ -21,6 +21,7 @@ | .planning/UPSTREAM-COMPARISON-2026-09-05.md | CURRENT | 2026-09-05 | Seventeen pinned upstream snapshots, source comparisons and bounded proposals; no head-to-head runtime benchmark. | | .planning/UPSTREAM-SNAPSHOTS-2026-09-05.json | CURRENT | 2026-09-05 | Origins, baselines, exact SHAs, branches, license labels and explicit inspection coverage. | | docs/development.md | CURRENT | 2026-09-05 | Scoped verification, public lifecycle demonstration, and deeper agent/GitNexus guidance. | +| docs/project-projections.md | CURRENT | 2026-10-22 | Default-off AO-3 source configuration, continuity, budgets and rollback; installation and real M2 remain pending. | | .planning/DREAMING-SOURCE-STEWARD-V1.md | CURRENT | 2026-09-05 | Dated deployed v1 checkpoint; its receipt contract is superseded for the candidate by USEFUL-SELECTION.md, without retroactive runtime activation. | | .planning/ENGINEERING-CONSOLIDATION-2026-09-05.md | CURRENT | 2026-09-05 | Bounded implementation evidence and remaining measured gaps, subordinate to ROADMAP.md. | | CHANGELOG.md | CURRENT | 2026-09-05 | Historical releases plus local 4.8.7 deployment; distinguishes fleet reconnection and public publication. | diff --git a/docs/generated/release-truth.json b/docs/generated/release-truth.json index 68f3f1b0..742daa48 100644 --- a/docs/generated/release-truth.json +++ b/docs/generated/release-truth.json @@ -144,7 +144,7 @@ "console_entrypoints": 9, "mcp_tools": 51, "ops_cli_commands": 5, - "pytest_test_functions": 5782 + "pytest_test_functions": 5923 }, "feature_profile_matrix": { "capture_hook": [ diff --git a/docs/generated/release-truth.md b/docs/generated/release-truth.md index 42a7de62..5cc1554c 100644 --- a/docs/generated/release-truth.md +++ b/docs/generated/release-truth.md @@ -13,7 +13,7 @@ Do not edit this file by hand. Run `python scripts/generate_release_truth.py`. - Main CLI commands: **127** - Operations CLI commands: **5** - Console entrypoints: **9** -- Pytest source test functions: **5782** +- Pytest source test functions: **5923** ## MCP tools diff --git a/docs/project-projections.md b/docs/project-projections.md new file mode 100644 index 00000000..aace3ed8 --- /dev/null +++ b/docs/project-projections.md @@ -0,0 +1,324 @@ + + + + +# Project projections + +The deterministic project reader serves confirmed cited decisions, constraints, +purpose and gotchas, plus an authorized filtered ledger export. SQLite remains +authoritative. Cards and generated Markdown never become claim evidence. + +The implementation follows the complete AO-3 design in +`artifacts/inputs/2026-10-22-AO-3-PROJECTION-DESIGN-v3.md` and its round-two +review. It changes package source and hook templates only. Installation, +external Cockpit integration, activation and the independent real three-repository +M2 probe remain separate owner actions. Fixture success is not installed M2 +acceptance. Original operator notes and GitNexus symbols remain v1.1 work. + +## Configuration + +All five switches default off. A child switch cannot bypass the master gate: + +| Environment variable | Purpose | +|---|---| +| `MEMORYMASTER_PROJECT_PROJECTION_ENABLED` | Master build, serve and state gate | +| `MEMORYMASTER_PROJECT_PROJECTION_REFRESH` | Separate deterministic beat phase | +| `MEMORYMASTER_PROJECT_PROJECTION_SESSION_START` | Shared project hook delivery | +| `MEMORYMASTER_PROJECT_SESSION_DELTA` | Start baselines and explicit end receipts | +| `MEMORYMASTER_PROJECT_PROJECTION_COCKPIT` | Reserved external consumer activation | + +`MEMORYMASTER_PROJECT_PROJECTION_PROJECTS` defaults to an empty JSON list. +Each of at most three mappings supplies `scope`, `context_key`, `project_root`, +`db_path`, explicit `tenant_id` (JSON null means only the unset legacy tenant), +optional `aliases`, `ledger_repo` and `ledger_filtered: true`. The latter asserts +that the lane supplied an authorized visibility-filtered export. Blind/private +cards are still excluded. Unknown mapping or unavailable/malformed ledger data +cannot mean zero open cards. + +Approved absolute roots are supplied through +`MEMORYMASTER_PROJECT_PROJECTION_DIR`, `MEMORYMASTER_PROJECT_SESSION_STATE_DIR` +and `MEMORYMASTER_PROJECT_LEDGER_TASKS_DIR`. There are no home-directory output +defaults. `MEMORYMASTER_PROJECT_PROJECTION_EPOCH` is a required fresh activation +token; retired-epoch cache and continuity state are ignored. Context keys and +epochs are bounded opaque names, not paths. The repository owner, strict +transcript resolver and generic scope resolver must agree before hook delivery. + +Manual commands branch before service construction or migrations: + +```text +python -m memorymaster.profile project-build --help +python -m memorymaster.profile project-read --help +``` + +Both require explicit database, scope, tenant, context, epoch, repository, +projection and state roots. Use `--tenant legacy-null` only for NULL tenancy. +Ledger arguments identify the filtered input. The master flag is still required. +These commands do not enable automatic refresh or consumers. + +## Integrity and budgets + +`project-.md` and its JSON manifest live below the explicit context root. +One kernel lock serializes publishers, unique temporary files are flushed and +replaced, and readers verify the pair, generation, byte counts, item spans and +hashes. A torn pair is unavailable. Rehydration uses a read-only snapshot, exact +tenant/scope checks and bounded primary-key support reads. A changed claim +withholds only its dependent bullets. A changed still-open card is rendered +from its current authorized record, without a false trimmed-card warning. + +The complete projection file, including its marker pair, is at most 2,200 UTF-8 +bytes. The delta is at most 500 bytes including labels, references and counts. +SessionStart wraps the marker-free shared payload, delta and fitting complete +profile lines in one marker pair and checks the final 3,000-character ceiling. +It never slices a validated project payload. Cards and the latest three +decisions reserve space before optional prose. Mandatory omissions visibly +block M2 readiness. The public manifest excludes protected source coverage. + +Builder snapshots cap all watched project rows at 512, citations at 2,048, +cards at 256, configured ledger records at 1,024 and source bytes at 8 MiB. +The all-status watch cap is deliberately stricter than counting eligible +claims: candidates and unsupported rows may later become eligible. An exceeded +cap is partial. Reader support limits are 64 claims, 256 citations and 2 MiB +shared source bytes, with a 25 ms SQLite busy timeout and cooperative 100 ms +SQL deadline. OS filesystem stalls are not hard real-time bounded. + +## Continuity and cheap refresh + +Session baselines contain fingerprints and an internal bounded scan fence, +never source prose or raw session tokens. The fence watches existing source +IDs, including ineligible rows, and the transactional change cursor of +migration 0029 (an epoch plus a per-write sequence, stamped inside the same +SQLite transaction as every claim and citation write). Incremental starts read +only the sequence range above the baseline cursor plus the temporal boundaries +crossed since the read, then deduplicate before loading payloads. This catches +same-second writes, a previously unsupported claim gaining a citation, and +old-timestamp writes committed after the prior read, with constant SQL work +when nothing changed (regression: `test_unchanged_other_projects_do_not_force_incremental_source_scan`). +Historical note: an earlier arm also scanned IDs above the greatest observed ID; +it re-traversed unrelated projects' rows on every unchanged start and was removed. +Incomplete or older fence-less metadata is partial, not a false no-change result. + +Delta selection unions differences from the selected prior start baseline and +the active projection's coverage. End events keep the start snapshot; they do +not acknowledge changes at end time. Missing projection coverage still permits +the prior-baseline comparison, labeled partial. After twelve hours an unclosed +active start can supply the newest stale baseline, explicitly labeled +`prior_session_unclosed`; inactivity never fabricates an end receipt. + +One atomic `sessions.json` per context/project retains at most sixteen active +starts, four completed starts and thirty-six deduplicated baseline blobs, each +at most 128 KiB, in a file capped at 8 MiB. Starts collect sources outside the +state lock and validate the sequence when reacquiring it. Duplicate starts +preserve their baseline. Replayed compacted ends, future ends, malformed state, +capacity exhaustion and lock/write failures cannot advance the latest-end +pointer. End recording precedes optional capture or distillation. Library +`run` never invents lifecycle receipts; the CLI requires both `--session-id` +and `--end-at` to record an explicit observed end. + +Refresh is separately gated and runs on enabled beat outcomes including SKIP, +capture-only and worker-busy. It checks SQLite data version, database/WAL and +ledger metadata before scanning; a bounded manifest hint survives scheduler +process restarts. Temporal boundaries defeat the cheap cache. An unchanged +SKIP does not rewrite Markdown or `checked_at`; successful non-SKIP validation +may update manifest freshness without rewriting Markdown. Changes rebuild. +Source failure or budget/lock refusal leaves the successful watermark intact. +Refresh rotates at most three pilot contexts under a cooperative 750 ms tick +budget. A validation older than sixty minutes is withheld pending refresh. + +## Activation and rollback + +Activation requires the lane's Argon source review and specific installation +and consumer-change authority. Install package/templates with every switch off, +configure approved contexts/roots and a new epoch, inspect manual builds first, +then separately authorize consumers, real M2 and automatic refresh. Existing +capture, Dreaming application and user-profile approvals remain independent. +The operational check reports cache/readiness and bounded receipt health. It +cannot substitute a healthy cache for a real independent installed M2 receipt. + +For rollback: + +1. Remove the affected allowlist mapping (or all pilot mappings). Set refresh, + SessionStart, delta and Cockpit switches to `0`, then set the master to `0` + in each affected environment. Record the retired epoch and sanitized reason + on the existing build card. +2. Reload or restart only affected long-lived consumers/jobs as separately + authorized. Fresh hooks read the changed environment on their next call. + Start a fresh conversation to verify rollback; previously injected context + cannot be retracted from an active conversation. +3. Verify the fresh hook uses its legacy branch, external Cockpit hides its + generated section, beat does no projection source/cache work and SessionEnd + writes no project continuity state. Disabled gates return before cache or + source reads. Do not disable unrelated capture/profile/beat features. +4. Retain derived files for inspection. Do not delete claims, restore a database + or ingest generated output. Keep the shipped span guard enforced. A later + activation uses a new epoch and fresh baselines. + +## Local verification record + +Red-first checks demonstrated missing modules and the new behavioral failures +before implementation: raw generated-span intake/dedup and HTML capture; +full-turn transcript filtering; hook lifecycle ordering; per-bullet card edits; +reserved UTF-8 budgets; malformed/out-of-order session state; missing projection +continuity; unresolved scope refusal; scheduler restart hints; same-second and +late committed writes; first citations; and equal-stat ledger rewrites. + +The explicit ten-thousand-session stress test passed with bounded deduplicated +state. A fixture M2 exercise uses three disposable repositories and template +hooks with the shared reader, with no independent SQL/model fallback. External +Cockpit and the real installed probe are not represented as completed. + +Windows sandbox directory creation required a test-launcher-only wrapper that +uses mode `0777` for disposable pytest directories. Production code and shared +pytest fixtures were not changed for that issue. The first broad service +coverage attempt was interrupted during provider initialization; affected +ingest cases passed with explicit hash embeddings. The first non-ML attempt +was interrupted before failures to give the ten-thousand-session stress test +its explicit 180-second timeout; it had no test failures at interruption. +Final suite counts and commands are recorded in the task close report. + +### Resumed source checks, 2026-10-22 + +At the end of the previous resume, source acceptance was **partial** pending the round-two incremental-read +criterion. `load_project_snapshot` unions every watched claim ID into the +SessionStart lookup; `_read_claims` then reloads those rows and their citations. +The query is bounded and uses primary-key lookups, but does not reuse unchanged +coverage without materializing its sources. A disposable unchanged-source +probe with three watched claims observed three rehydrated rows and three +rehydrated citations. `test_session_start_baseline_never_full_source_scan` +checks the query arguments only and does not establish incremental cost. + +Simply removing watched IDs would regress the passing citation-only, +same-second, first-citation and old-timestamp late-commit tests. A bounded SQL +fingerprint comparison could reduce Python materialization while still reading +all watched source fingerprints. Strict change-only reads need a trustworthy +mutation sequence or another approved design. The lane's Argon review must +resolve that difference before claiming the incremental criterion or activating +the pilot; no journal, migration or activation was added in this resume. + +The resume also found and fixed pinned decisions incorrectly sorting behind +older unpinned decisions. The new +`test_latest_decisions_ignore_pin_priority` first failed with observed IDs +`[11, 10, 9]` instead of `[14, 13, 12]`; the complete projection file then passed +all twelve tests. Pins still order constraints, while decisions use date and ID. + +Freshly executed checks, one test file per Python process: + +| Test file (`tests/`) | Passed | Skipped | +|---|---:|---:| +| `test_project_projection.py` | 12 | 0 | +| `test_project_repository.py` | 22 | 0 | +| `test_project_sessions.py` | 13 | 0 | +| `test_project_runtime.py` | 8 | 0 | +| `test_project_integration.py` | 5 | 0 | +| `test_generated_artifacts.py` | 3 | 0 | +| `test_compiled_profile_surfaces.py` | 6 | 0 | +| `test_profile_config_defaults.py` | 2 | 0 | +| `test_compiled_profile.py` | 9 | 0 | +| `test_compiled_profile_claim_supports.py` | 50 | 0 | +| `test_intake_policy.py` | 22 | 0 | +| `test_service.py` | 6 | 0 | +| `test_capture_adapters.py` | 24 | 1 | +| `test_public_v1.py` | 14 | 0 | +| `test_dreaming_self_observation.py` | 35 | 0 | +| `test_session_end_capture_hook.py` | 2 | 0 | +| `test_session_end_ingest.py` | 4 | 0 | +| `test_scheduled_task_runtime.py` | 7 | 0 | +| `test_operational_review.py` | 16 | 0 | +| `test_public_demo.py` | 2 | 0 | +| `test_scope_utils.py` | 17 | 0 | +| **Total** | **279** | **1** | + +Runs used `WEZBRIDGE_OPERATOR_ALLOW_HEAVY=1`, disposable workspace basetemps +and the local directory-mode wrapper described above. Service-backed suites +used `MEMORYMASTER_EMBEDDING_PROVIDER=hash`; default temporary-directory roots +were redirected into the workspace. The sole skip requires Windows symlink +privilege. The ten-thousand-session case passed in the complete session file +(125 seconds). No full-suite or installed/live acceptance is claimed. + +The first projection attempt had eleven fixture setup errors because its +basetemp parent did not exist; the next exposed the sandbox's `0700` directory +access failure. Creating the parent and using the disposable-only wrapper +resolved those harness errors. The first standalone incremental probe imported +the installed package because its script path omitted the checkout; explicitly +prepending the checkout resolved that harness error before the observation +above. These were retained as failed attempts, not product regressions. + +Ruff passed on all thirty-three changed Python files. Release truth was +regenerated for the new regression and `--check` passed. Contract witness +passed against `origin/main` (seventeen code files, sixteen test files, no +orphan witnesses). GitNexus reported `INDEX_MISSING`; deferred tool discovery +exposed neither GitNexus nor Serena. Scoped source references and the reviewed +diff covered the builder's CLI/refresh callers before committing. Branch +freshness passed with zero commits behind `origin/main`. + +## Round-three incremental collection (T-1703) + +The incremental source criterion now passes locally. The first red run of +`test_session_start_incremental_counts_only_changed_rows_and_citations` +failed all three cases: forty known claims caused forty claim and citation +loads even with zero, one or three changes. The corrected cases load exactly +zero, one or three claim rows and their citations, including overlapping claim +and citation edits, while retaining unchanged support hashes. Unchanged rows +from the same second are also excluded. SQL query plans assert primary-key +payload lookups and indexed citation lookups without scanning the source tables. + +The previous timestamp-only gap demonstrated the need for versioned migration +29: a singleton epoch/sequence clock and one latest-change row per claim, +including deletion tombstones. Transactional triggers track claim and citation +inserts, updates, deletes and identity moves without storing source prose. +This is bounded per claim identity rather than an append-only event history. +Indexed sequence changes are unioned with crossed temporal-boundary ranges, +then deduplicated before loading payloads (the earlier `updated_at` and new-ID +arms described in older notes were superseded by the transactional cursor). +The same read-only SQLite transaction captures the clock, selected IDs, claims +and citations. Coverage carries unchanged manifest support hashes and removes +changed identities before adding currently eligible supports. + +The source-load cost is O(N + R): N unique changed claims and their citations, +plus R supports needed for prior-session delta prose when a refreshed +projection already includes those changes. Lazy delta loading stops at the +500-byte output limit or the first unavailable support, with at most one +additional candidate. A forty-claim refreshed-projection regression counts +those loads against the rendered delta supports. The shared projection keeps +its 2,200-byte cap and delta entries retain exact support hashes and spans. +Coverage and bounded session-state serialization still operate on compact +metadata, and configured ledger reads retain their existing bounds. + +Migration 29 is SQLite-only and PostgreSQL fails closed. Writable store +initialization installs it, including convergence after a standalone migration +runner stamped an empty database. SessionStart never initializes or migrates +the store. Missing, foreign or rewound cursors withhold the baseline without +falling back to a source scan; old projections require a fresh generation after +schema initialization. The migration and initialization checks used disposable +databases only, including backup/restore and late WAL commits. No authoritative +database or installed service was changed; activation and independent Argon/M2 +acceptance remain separate lane actions. + +Retained failed evidence: the first repository integration run had two failures +from tests that assumed an empty coverage could bootstrap an incremental read +and that payload SQL contained the old watched-ID union. Those witnesses now +check temporal activation/expiration with known coverage and the indexed +selection followed by a deduplicated payload lookup. Cursor tests first exposed +an untracked old primary key on identity moves and a bare-runner initialization +gap; both received failing regressions before their fixes. + +Fresh round-three verification used one test file per Python process and the +same workspace-only temporary-directory launcher as the previous resume: + +| Test file (`tests/`) | Passed | +|---|---:| +| `test_project_projection.py` | 16 | +| `test_project_repository.py` | 31 | +| `test_project_sessions.py` | 14 | +| `test_project_runtime.py` | 8 | +| `test_project_integration.py` | 5 | +| `test_project_source_cursor.py` | 13 | +| `test_public_demo.py` | 2 | +| `test_migrations.py` | 18 | +| `test_initdb_fastpath.py` | 6 | +| **Total** | **113** | + +Ruff passed on all eight changed Python files. Release truth was regenerated +and `--check` passed. Contract witness passed against `origin/main`; source +navigation and final change-scope review used live references and Git diffs +because this checkout still has no GitNexus index or exposed navigation tools. +No push, merge, installed-hook edit or live database check belongs to this round. diff --git a/memorymaster/capture/adapters.py b/memorymaster/capture/adapters.py index bd571ed6..c7b9caaa 100644 --- a/memorymaster/capture/adapters.py +++ b/memorymaster/capture/adapters.py @@ -8,13 +8,16 @@ import mimetypes import os import re +from bisect import bisect_right from dataclasses import dataclass +from html import unescape from html.parser import HTMLParser from pathlib import Path from typing import Protocol from urllib.parse import urlsplit from zipfile import BadZipFile, ZipFile +from memorymaster.core.generated_artifacts import SourceSpan, strip_generated_spans from memorymaster.core.security import ( sanitize_persisted_text, validate_persisted_metadata, @@ -72,6 +75,8 @@ class CaptureEnvelope: producer_session_hash: str | None = None producer_turn_id: str | None = None producer_metadata: tuple[tuple[str, str], ...] = () + retained_spans: tuple[SourceSpan, ...] = () + excluded_spans: tuple[tuple[int, int], ...] = () class CaptureAdapter(Protocol): @@ -83,21 +88,115 @@ class _VisibleTextParser(HTMLParser): def __init__(self) -> None: super().__init__(convert_charrefs=True) self.hidden_depth = 0 + self.hidden_tags: list[str] = [] self.parts: list[str] = [] def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None: if tag.lower() in {"script", "style", "noscript", "template"}: - self.hidden_depth += 1 + self.hidden_tags.append(tag.lower()) + self.hidden_depth = len(self.hidden_tags) def handle_endtag(self, tag: str) -> None: - if tag.lower() in {"script", "style", "noscript", "template"} and self.hidden_depth: - self.hidden_depth -= 1 + if self.hidden_tags and tag.lower() == self.hidden_tags[-1]: + self.hidden_tags.pop() + self.hidden_depth = len(self.hidden_tags) def handle_data(self, data: str) -> None: if not self.hidden_depth and data.strip(): self.parts.append(data.strip()) +class _RetainedHTMLParser(_VisibleTextParser): + """Parse masked HTML once; map visible data to original retained ranges.""" + + def __init__(self, source: str, spans: tuple[SourceSpan, ...], excluded: tuple[tuple[int, int], ...]) -> None: + super().__init__() + # Decode entities after mapping their original source ranges. + self.convert_charrefs = False + self.source = source + self.spans = spans + self.ends = [span.end for span in spans] + self.excluded = excluded + self.excluded_ends = [end for _, end in excluded] + self.line_starts = [0, *(match.end() for match in re.finditer("\n", source))] + self.mapped: list[list[tuple[int, int, str]]] = [[] for _ in spans] + + def _check_token(self, length: int) -> None: + line, column = self.getpos() + start = self.line_starts[line - 1] + column + index = bisect_right(self.excluded_ends, start) + if index < len(self.excluded) and self.excluded[index][0] < start + length: + raise CaptureRejected("ambiguous_generated_provenance", "HTML token crosses excluded generated content.") + + def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None: + self._check_token(len(self.get_starttag_text())) + super().handle_starttag(tag, attrs) + + def handle_endtag(self, tag: str) -> None: + line, column = self.getpos() + start = self.line_starts[line - 1] + column + self._check_token(self.source.index(">", start) + 1 - start) + super().handle_endtag(tag) + + def handle_comment(self, data: str) -> None: + self._check_token(len(data) + 7) + + def handle_decl(self, decl: str) -> None: + self._check_token(len(decl) + 3) + + def handle_pi(self, data: str) -> None: + self._check_token(len(data) + 2) + + def unknown_decl(self, data: str) -> None: + self._check_token(len(data) + 5) + + def _record(self, data: str, raw_length: int, *, entity: bool = False) -> None: + if self.hidden_depth: + return + line, column = self.getpos() + start = self.line_starts[line - 1] + column + end = start + raw_length + index = bisect_right(self.ends, start) + while index < len(self.spans) and self.spans[index].start < end: + span = self.spans[index] + left, right = max(start, span.start), min(end, span.end) + if entity and (left != start or right != end): + raise CaptureRejected("ambiguous_generated_provenance", "Split HTML entity cannot be attributed safely.") + value = data if entity else data[left - start:right - start] + self.mapped[index].append((left, right, value)) + index += 1 + + def handle_data(self, data: str) -> None: + self._record(data, len(data)) + + def _entity(self, name: str, prefix: str) -> None: + line, column = self.getpos() + start = self.line_starts[line - 1] + column + raw = prefix + name + if self.source[start + len(raw):start + len(raw) + 1] == ";": + raw += ";" + self._record(unescape(raw), len(raw), entity=True) + + def handle_entityref(self, name: str) -> None: + self._entity(name, "&") + + def handle_charref(self, name: str) -> None: + self._entity(name, "&#") + + def values(self) -> list[str]: + result = [] + for parts in self.mapped: + chunks = [] + previous_end = None + for start, end, value in parts: + if chunks and previous_end != start: + chunks.append("\n") + chunks.append(value) + previous_end = end + result.append("".join(chunks).strip()) + return result + + def _digest(payload: bytes) -> str: return hashlib.sha256(payload).hexdigest() @@ -106,11 +205,104 @@ def _bounded_text(text: str) -> tuple[str, tuple[str, ...]]: payload = text.encode("utf-8") if len(payload) > EXTRACTED_TEXT_LIMIT: raise CaptureRejected("extracted_text_too_large", "Extracted text exceeds 2 MiB.") - sanitized, findings = sanitize_persisted_text(text) - warnings = ("sensitive_content_redacted",) if findings else () + retained = strip_generated_spans(text) + if retained.excluded_spans and not retained.spans: + raise CaptureRejected("generated_artifact", "Generated-only content cannot become capture evidence.") + sanitized, findings = sanitize_persisted_text(retained.text) + warnings = (("generated_spans_excluded",) if retained.excluded_spans else ()) + (("sensitive_content_redacted",) if findings else ()) return sanitized, warnings +def _masked_html_source(text: str, excluded: tuple[tuple[int, int], ...]) -> str: + """Make generated bytes inert without moving source offsets or line starts.""" + retained_chunks = [] + position = 0 + for start, end in excluded: + retained_chunks.append(text[position:start]) + position = end + retained_chunks.append(text[position:]) + retained = "".join(retained_chunks) + retained_end = 0 + chunks = [] + position = 0 + lexer = HTMLParser(convert_charrefs=False) + for start, end in excluded: + prefix = text[position:start] + retained_end += len(prefix) + lexer.feed(prefix) + pending = lexer.rawdata + # Whitespace must not complete an authored tag/entity or leave an + # incomplete token to absorb the next retained span. Refuse ambiguity + # rather than inventing a tag or joining independently authored bytes. + if lexer.cdata_elem: + # In script/style raw text only the matching closing tag is HTML + # syntax. Test both retained sides with the parser's current closing + # pattern: a possible prefix alone is harmless string data. This + # lookahead also catches tokens split across several exclusions; + # the compact text is never fed to the offset-preserving parser. + tail_start = retained.rfind("<", 0, retained_end) + closing = lexer.interesting.match(retained, tail_start) if tail_start >= 0 else None + # A match that ends exactly at the boundary is a closing-tag name whose terminator + # lies past the excluded bytes (hardened html.parser matches ''), so it crosses as well. + ambiguous = bool(closing and closing.end() >= retained_end) + else: + ambiguous = bool( + re.search(r"<(?:/?[A-Za-z][^<>]*|/?|![^<>]*|\?[^<>]*)$", pending) + or pending.rfind("") + or re.search(r"&(?:\#(?:[xX][0-9a-fA-F]*|[0-9]*)|[A-Za-z][A-Za-z0-9]*)?$", pending) + ) + if ambiguous: + raise CaptureRejected("ambiguous_generated_provenance", "Incomplete HTML token touches excluded generated content.") + chunks.append(prefix) + # Inside script/style raw text a run of spaces after a retained " tuple[str, tuple[str, ...], tuple[SourceSpan, ...], tuple[tuple[int, int], ...]]: + """Filter raw input before parsers erase the generated marker comments.""" + clean = strip_generated_spans(text) + if clean.excluded_spans and not clean.spans: + raise CaptureRejected("generated_artifact", "Generated-only content cannot become capture evidence.") + spans: list[SourceSpan] = [] + warning_codes: list[str] = ["generated_spans_excluded"] if clean.excluded_spans else [] + html_values = [] + if html: + masked = _masked_html_source(text, clean.excluded_spans) + parser = _RetainedHTMLParser(masked, clean.spans, clean.excluded_spans) + parser.feed(masked) + parser.close() + html_values = parser.values() + for index, span in enumerate(clean.spans): + if span.start == 0 and span.text.startswith("\ufeff"): + span = SourceSpan(1, span.end, span.text[1:]) + value = html_values[index].lstrip("\ufeff") if html else span.text + if html and strip_generated_spans(value).excluded_spans: + # HTML entity decoding changes offsets. Without a reversible source + # map, splitting decoded text would fabricate original attribution. + raise CaptureRejected( + "ambiguous_generated_provenance", "Decoded HTML generated markers cannot be attributed safely." + ) + bounded, warnings = _bounded_text(value) + warning_codes.extend(warnings) + if bounded.strip(): + spans.append(SourceSpan(span.start, span.end, bounded)) + if clean.excluded_spans and not spans: + raise CaptureRejected("generated_artifact", "Generated-only content cannot become capture evidence.") + if sum(len(span.text.encode("utf-8")) for span in spans) + max(0, len(spans) - 1) > EXTRACTED_TEXT_LIMIT: + raise CaptureRejected("extracted_text_too_large", "Extracted text exceeds 2 MiB.") + return "\n".join(span.text for span in spans), tuple(dict.fromkeys(warning_codes)), tuple(spans), clean.excluded_spans + + def _validate_source_uri(source_uri: str | None) -> str | None: if source_uri is None: return None @@ -199,7 +391,7 @@ def _decode_utf8(payload: bytes) -> str: if b"\x00" in payload: raise CaptureRejected("binary_masquerading", "Text document contains NUL bytes.") try: - return payload.decode("utf-8-sig") + return payload.decode("utf-8") except UnicodeDecodeError as exc: raise CaptureRejected("invalid_utf8", "Text and Markdown must be UTF-8.") from exc @@ -285,7 +477,7 @@ def capture(self) -> CaptureEnvelope: raw = self.text.encode("utf-8") if len(raw) > INLINE_LIMIT: raise CaptureRejected("inline_text_too_large", "Inline text exceeds 2 MiB.") - text, warnings = _bounded_text(self.text) + text, warnings, spans, excluded = _retained_text(self.text) uri = _validate_source_uri(self.source_uri) return CaptureEnvelope( source_kind="inline", @@ -297,6 +489,8 @@ def capture(self) -> CaptureEnvelope: source_uri=uri, evidence_type="text", warning_codes=warnings, + retained_spans=spans, + excluded_spans=excluded, ) @@ -350,7 +544,7 @@ def _envelope( evidence_type: str | None, provider: str | None = None, ) -> CaptureEnvelope: - bounded, warnings = _bounded_text(text) if text is not None else (None, ()) + bounded, warnings, spans, excluded = _retained_text(text, html=mime == "text/html") if text is not None else (None, (), (), ()) return CaptureEnvelope( source_kind="file", content_hash=_digest(payload), @@ -363,6 +557,8 @@ def _envelope( provider_kind=provider, warning_codes=warnings, resolved_path=str(resolved), + retained_spans=spans, + excluded_spans=excluded, ) def capture(self) -> CaptureEnvelope: @@ -374,7 +570,7 @@ def capture(self) -> CaptureEnvelope: if extension in _TEXT_EXTENSIONS: return self._envelope(resolved, locator, payload, "text/plain", _decode_utf8(payload), "text") if extension in _HTML_EXTENSIONS: - return self._envelope(resolved, locator, payload, "text/html", _html_text(payload), "text") + return self._envelope(resolved, locator, payload, "text/html", _decode_utf8(payload), "text") if extension == ".pdf": return self._envelope(resolved, locator, payload, "application/pdf", _pdf_text(payload), "text") if extension == ".docx": diff --git a/memorymaster/config_templates/hooks/memorymaster-session-end.py b/memorymaster/config_templates/hooks/memorymaster-session-end.py index 049431f9..2fa8f57e 100644 --- a/memorymaster/config_templates/hooks/memorymaster-session-end.py +++ b/memorymaster/config_templates/hooks/memorymaster-session-end.py @@ -2,9 +2,8 @@ import json import os +import sqlite3 import sys -import tempfile -from pathlib import Path PROJECT_ROOT = "__MEMORYMASTER_PROJECT_ROOT__" DB_PATH = os.path.join(PROJECT_ROOT, "memorymaster.db") @@ -19,12 +18,26 @@ def _enabled(name, default=True): def main(): - if not _enabled("MEMORYMASTER_SESSION_END_CAPTURE", True): - return try: data = json.loads(sys.stdin.read() or "{}") + if not isinstance(data, dict): + data = {} except Exception: data = {} + try: + from memorymaster.profile.project_runtime import record_session_end + + record_session_end(data) + except (ImportError, OSError, RuntimeError, ValueError, sqlite3.Error) as exc: + # Report only types: exception messages can carry session text, paths, + # identities or credentials. Capture has its own activation decision. + cause = exc.__cause__ or exc.__context__ or exc + print( + f"[MemoryMaster] session-end receipt error: type={type(exc).__name__} cause={type(cause).__name__}", + file=sys.stderr, + ) + if not _enabled("MEMORYMASTER_SESSION_END_CAPTURE", True): + return transcript_path = data.get("transcript_path", "") session_id = str(data.get("session_id") or "unknown") cwd = str(data.get("cwd") or PROJECT_ROOT) @@ -40,12 +53,11 @@ def main(): reservation = ledger.reserve_llm(provider, session_id, "session-end-distill") if reservation is None: return - temp_path = "" try: - with tempfile.NamedTemporaryFile("w", encoding="utf-8", suffix=".jsonl", delete=False) as handle: - handle.write(chunk.text) - temp_path = handle.name - run(DB_PATH, temp_path, source_agent="session-end-hook", cwd=cwd) + # Reconstruct structural provenance from the chronological source, + # but select evidence only within this cursor's complete records. + run(DB_PATH, transcript_path, source_agent="session-end-hook", cwd=cwd, + start_offset=chunk.start_offset, end_offset=chunk.end_offset) ledger.finish_llm(reservation, input_bytes=len(chunk.text.encode("utf-8")), output_bytes=0, outcome="ok") ledger.commit_cursor(chunk) except Exception: # swallow-ok: hook quiet-by-contract; `capture_usage.outcome` @@ -53,9 +65,6 @@ def main(): # Guardar la causa pide una columna nueva en capture_usage, que es otro # subsistema. Deuda anotada, no escondida. ledger.finish_llm(reservation, input_bytes=len(chunk.text.encode("utf-8")), output_bytes=0, outcome="error") - finally: - if temp_path: - Path(temp_path).unlink(missing_ok=True) except Exception: return diff --git a/memorymaster/config_templates/hooks/memorymaster-session-start.py b/memorymaster/config_templates/hooks/memorymaster-session-start.py index 71d7e1d3..102177b7 100644 --- a/memorymaster/config_templates/hooks/memorymaster-session-start.py +++ b/memorymaster/config_templates/hooks/memorymaster-session-start.py @@ -365,6 +365,24 @@ def main(): except Exception: pass + # Read the switch without the package: when the consumer is enabled, even an import + # failure must withhold legacy delivery; when it is off, the hook degrades as before. + project_enabled = os.environ.get("MEMORYMASTER_PROJECT_PROJECTION_ENABLED", "").strip().lower() in {"1", "true", "yes", "on"} + try: + from memorymaster.profile.project_runtime import session_start_context + + context = session_start_context(data, profile_loader=_load_compiled_profile) + if context is not None: + sys.stdout.write(json.dumps({"hookSpecificOutput": { + "hookEventName": "SessionStart", "additionalContext": context}})) + sys.stdout.flush() + sys.exit(0) + except Exception as exc: + # An enabled project consumer must never fall through to less restrictive legacy SQL. + if project_enabled: + _log("project-consumer-error", cause=type(exc).__name__) + sys.exit(0) + if not DB_PATH.exists(): _log("skip", reason="db-missing") sys.exit(0) diff --git a/memorymaster/core/generated_artifacts.py b/memorymaster/core/generated_artifacts.py new file mode 100644 index 00000000..d6d19bfa --- /dev/null +++ b/memorymaster/core/generated_artifacts.py @@ -0,0 +1,67 @@ +"""Recognize generated envelopes without turning separated source spans into evidence.""" + +from __future__ import annotations + +import re +from dataclasses import dataclass + +GENERATED_OPENER = "" +GENERATED_CLOSER = "" +GENERATED_MARKER = GENERATED_OPENER +_MARKER = re.compile( + r"", re.I, +) + + +@dataclass(frozen=True) +class SourceSpan: + start: int + end: int + text: str + + +@dataclass(frozen=True) +class GeneratedText: + text: str + spans: tuple[SourceSpan, ...] + excluded_spans: tuple[tuple[int, int], ...] = () + + +def contains_generated_marker(text: str) -> bool: + return bool(_MARKER.search(text)) + + +def strip_generated_spans(text: str) -> GeneratedText: + """Keep original character offsets; unclosed/legacy envelopes own their suffix. + + ``text`` is a display convenience. Evidence consumers use ``spans`` rather + than asserting that text across a removed range was one authored statement. + Stray closers are excluded too, so they cannot survive as intake markers. + """ + removed: list[tuple[int, int]] = [] + depth = 0 + start = 0 + for marker in _MARKER.finditer(text): + if marker.group("close"): + if depth: + depth -= 1 + if depth == 0: + removed.append((start, marker.end())) + else: + removed.append(marker.span()) + else: + if depth == 0: + start = marker.start() + depth += 1 + if depth: + removed.append((start, len(text))) + spans: list[SourceSpan] = [] + position = 0 + for start, end in removed: + if position < start: + spans.append(SourceSpan(position, start, text[position:start])) + position = end + if position < len(text): + spans.append(SourceSpan(position, len(text), text[position:])) + spans = [span for span in spans if span.text.strip("\ufeff \t\r\n")] + return GeneratedText("\n".join(span.text for span in spans), tuple(spans), tuple(removed)) diff --git a/memorymaster/core/intake_policy.py b/memorymaster/core/intake_policy.py index 30d60077..bd08af3b 100644 --- a/memorymaster/core/intake_policy.py +++ b/memorymaster/core/intake_policy.py @@ -265,6 +265,13 @@ def evaluate_intake( the Nth+1 claim of that batch is rejected. ``intake_batch_max`` overrides the configured ``max_per_stop`` for that batch only. """ + from memorymaster.core.generated_artifacts import contains_generated_marker + + if not text.strip() or any(contains_generated_marker(value) for value in (text, subject or "")): + raise IntakeRejected( + "Empty or generated content is not an attributable knowledge claim.", + rule="generated_artifact", reason="unattributed_content", + ) cfg = config or IntakePolicyConfig.from_env() moment = time.time() if now is None else now mutated: dict[str, object] = {} diff --git a/memorymaster/core/service.py b/memorymaster/core/service.py index 1475e27d..4c1c3380 100644 --- a/memorymaster/core/service.py +++ b/memorymaster/core/service.py @@ -486,6 +486,17 @@ def init_db(self) -> None: ) self.store.init_db() + @staticmethod + def _validate_dedup_lineage(claim: Claim) -> None: + """Old rows and race winners must satisfy the same intake boundary.""" + from memorymaster.core.generated_artifacts import contains_generated_marker + + fields = [claim.text, claim.subject, claim.predicate, claim.object_value] + fields.extend(value for citation in claim.citations + for value in (citation.source, citation.locator, citation.excerpt)) + if any(contains_generated_marker(value) for value in fields if value): + raise ValueError("Stored dedup claim has generated content or citation evidence.") + def _revive_archived_dedup_match(self, claim: Claim, source_agent: str | None) -> Claim: """Resurface an archived claim matched by ingest dedup. @@ -498,6 +509,7 @@ def _revive_archived_dedup_match(self, claim: Claim, source_agent: str | None) - review. Explicit forget/rollback intent blocks automatic revival. Non-archived matches pass through untouched (dedup unchanged). """ + self._validate_dedup_lineage(claim) if claim.status != "archived": return claim if self.store.latest_archive_reason(claim.id) in {"friendly_forget_claim", "ao7_rollback"}: @@ -517,6 +529,8 @@ def _revive_archived_dedup_match(self, claim: Claim, source_agent: str | None) - # Another writer touched the row between dedup and revival — # the freshest state wins; never fail the whole ingest for it. reloaded = self.store.get_claim(claim.id) + if reloaded is not None: + self._validate_dedup_lineage(reloaded) return reloaded if reloaded is not None else claim # getattr-guard: some paths build MemoryService via __new__ and skip # __init__, so the qdrant attribute may not exist. @@ -551,6 +565,23 @@ def ingest( outcome_receipt: dict[str, Any] | None = None, ) -> Claim: prepare_outcome_receipt(outcome_receipt) + from memorymaster.core.generated_artifacts import contains_generated_marker, strip_generated_spans + + retained = strip_generated_spans(text) + raw_fields = [subject, predicate, object_value] + raw_fields.extend(value for citation in citations for value in (citation.source, citation.locator, citation.excerpt)) + if any(contains_generated_marker(value) for value in raw_fields if value): + raise ValueError("Claim has generated structured fields or citation evidence.") + if retained.excluded_spans: + # A claim is atomic. Document capture can preserve several authored + # spans, but joining them here would invent a single assertion. + if len(retained.spans) != 1 or idempotency_key: + raise ValueError("Claim has generated content without atomic retained attribution.") + text = retained.spans[0].text.strip() + if any(value and value not in text for value in (subject, predicate, object_value)): + raise ValueError("Claim has generated content without structured attribution.") + if any(citation.excerpt and citation.excerpt not in text for citation in citations): + raise ValueError("Claim has generated content without retained citation attribution.") if not text.strip(): raise ValueError("Claim text cannot be empty.") if not citations: @@ -653,6 +684,7 @@ def ingest( visibility=visibility, source_agent=source_agent, ) if retired is not None: + self._validate_dedup_lineage(retired) fill_outcome_receipt(outcome_receipt, retired, created=False, persisted=self.store.ingest_creation_receipt(retired.id) if outcome_receipt is not None else None) observability.bump_claim_ingested(source_agent) diff --git a/memorymaster/dreaming/capture.py b/memorymaster/dreaming/capture.py index e65af3d9..5edd9b81 100644 --- a/memorymaster/dreaming/capture.py +++ b/memorymaster/dreaming/capture.py @@ -10,6 +10,7 @@ from typing import Any, Mapping from memorymaster.core.capture_control import CaptureLedger +from memorymaster.core.generated_artifacts import SourceSpan from memorymaster.core.security import redact_text from memorymaster.core.scope_utils import resolve_transcript_workspace_scope, scope_from_cwd from memorymaster.dreaming.ledger import DreamLedger @@ -35,7 +36,7 @@ def _text_parts(content: Any, allowed: set[str]) -> str: def _conversation_record( entry: dict[str, Any], provider: str, *, state: dict[str, Any] | None = None, -) -> tuple[str, str] | None: +) -> tuple[str, tuple[SourceSpan, ...]] | None: state = state if state is not None else {} payload = entry.get("payload") if provider == "codex" else entry.get("message", entry) payload = payload if isinstance(payload, dict) else entry @@ -73,7 +74,9 @@ def _conversation_record( "injected": bool(entry.get("hook_additional_context") or entry.get("isMeta") or payload.get("hook_additional_context")), "generated_session": bool(entry.get("backfill_generated") or entry.get("memorymaster_generated")), }) - return (normalized_role, clean.text) if normalized_role and clean.text else None + # The convenience joined text is never attributable evidence: a removed + # range can separate an authored prefix from an unrelated suffix. + return (normalized_role, clean.spans) if normalized_role and clean.spans else None def _sanitize(text: str) -> str: @@ -98,18 +101,21 @@ def parse_transcript_lines( record = _conversation_record(entry, provider.lower(), state=state) if record is None: continue - role, content = record - sanitized = _sanitize(content) - if not sanitized: - continue + role, spans = record timestamp = str(entry.get("timestamp") or _now()) - identity = (role, sanitized, timestamp) - if identity in seen: - continue - seen.add(identity) - raw_id = str(entry.get("uuid") or entry.get("id") or "|".join(identity)) - message_id = "dm-" + hashlib.sha256(raw_id.encode("utf-8", errors="replace")).hexdigest()[:20] - messages.append(DreamMessage(message_id, role, sanitized, timestamp)) + for span in spans: + sanitized = _sanitize(span.text) + if not sanitized: + continue + identity = (role, sanitized, timestamp) + if identity in seen: + continue + seen.add(identity) + raw_id = str(entry.get("uuid") or entry.get("id") or "|".join(identity)) + if len(spans) > 1: + raw_id += f":span:{span.start}:{span.end}" + message_id = "dm-" + hashlib.sha256(raw_id.encode("utf-8", errors="replace")).hexdigest()[:20] + messages.append(DreamMessage(message_id, role, sanitized, timestamp)) return messages diff --git a/memorymaster/dreaming/self_observation.py b/memorymaster/dreaming/self_observation.py index 5cb44127..8281d835 100644 --- a/memorymaster/dreaming/self_observation.py +++ b/memorymaster/dreaming/self_observation.py @@ -6,12 +6,7 @@ from dataclasses import dataclass from typing import Any, MutableMapping - -@dataclass(frozen=True) -class SourceSpan: - start: int - end: int - text: str +from memorymaster.core.generated_artifacts import SourceSpan, strip_generated_spans @dataclass(frozen=True) @@ -73,17 +68,18 @@ def strip_self_observation( return SanitizedTurn("", exclusion_reasons=("structural_context",), contaminated=True) if memory_tool: return SanitizedTurn("", exclusion_reasons=("memory_tool",), contaminated=True) - removed: list[tuple[int, int]] = [] + generated = strip_generated_spans(text) + removed: list[tuple[int, int]] = list(generated.excluded_spans) for match in _WRAPPER.finditer(text): # Persisted outputs are never followed; even an unmarked pointer may be # recalled content whose provenance was erased by the export. removed.append(match.span()) spans: list[SourceSpan] = [] position = 0 - for start, end in removed: + for start, end in sorted(removed): if position < start: spans.append(SourceSpan(position, start, text[position:start])) - position = end + position = max(position, end) if position < len(text): spans.append(SourceSpan(position, len(text), text[position:])) remaining = "\n".join(span.text for span in spans) @@ -94,8 +90,9 @@ def strip_self_observation( state["tool_contaminated"] = True return SanitizedTurn("", exclusion_reasons=("generated_memory",), contaminated=True) authored = remaining.strip() - if removed and not authored: + if removed and not authored.strip("\ufeff"): state["tool_contaminated"] = True + return SanitizedTurn("", exclusion_reasons=("generated_memory",), contaminated=True) if role == "user" and authored: state["tool_contaminated"] = False state["memory_tool_ids"] = [] diff --git a/memorymaster/operations/operational_review.py b/memorymaster/operations/operational_review.py index ca30c8fc..3e886644 100644 --- a/memorymaster/operations/operational_review.py +++ b/memorymaster/operations/operational_review.py @@ -85,6 +85,9 @@ class ReviewConfig: snapshot_dir: Path | None = None dream_log: Path | None = None steward_log: Path | None = None + # AO-3 uses explicit approved contexts; no database/home/root discovery. + project_projection_configs: tuple[Any, ...] = () + project_m2_receipt: Path | None = None def _configured_canaries(config: ReviewConfig) -> list[tuple[str, str]]: @@ -1003,11 +1006,83 @@ def progress_counts(bucket: dict[str, Any]) -> dict[str, int]: return ReviewResult("backfill_burndown", Verdict.PASS, detail, counts) +def check_project_projection(config: ReviewConfig) -> ReviewResult: + """Validate derived views and receipt health without publishing or source mutation.""" + from memorymaster.profile.project_runtime import PREFIX, active, load_project_configs + + if not active(PREFIX + "ENABLED"): + return ReviewResult("project_projection", Verdict.PASS, "inactive; all project consumers gated") + contexts = config.project_projection_configs or tuple(load_project_configs()) + if not contexts: + return ReviewResult("project_projection", Verdict.WARN, "pending: explicit allowlist/roots/epoch unavailable") + from memorymaster.profile.project_projection import read_manifest, read_project_projection + from memorymaster.profile.project_sessions import read_session_state + + counts = {"projects": len(contexts), "ready": 0, "bytes": 0, "active_sessions": 0, + "completed_sessions": 0, "baseline_blobs": 0, "m2_passed": 0} + diagnostics = [] + views = {} + failed = False + for context in contexts: + try: + manifest = read_manifest(context) + if manifest is None: + failed = True + diagnostics.append("invalid_or_missing_pair") + continue + checked = datetime.fromisoformat(manifest["checked_at"].replace("Z", "+00:00")) + age = (datetime.now(timezone.utc) - checked).total_seconds() + counts["oldest_check_seconds"] = max(counts.get("oldest_check_seconds", 0), int(age)) + view = read_project_projection(context) + if view is None: + diagnostics.append("stale_or_unavailable_view") + continue + counts["bytes"] += view.manifest["payload_bytes"] + if view.status == "ready": + counts["ready"] += 1 + else: + diagnostics.append("partial_or_m2_unready") + if age > 3600: + diagnostics.append("stale_source_validation") + state = read_session_state(context) + counts["active_sessions"] += len(state["active"]) + counts["completed_sessions"] += len(state["completed"]) + counts["baseline_blobs"] += len(state["baselines"]) + if active("MEMORYMASTER_PROJECT_SESSION_DELTA") and state["latest_end"] is None: + diagnostics.append("baseline_unavailable") + views[context.scope] = view.manifest["view_id"] + except (OSError, ValueError, KeyError, TypeError): + diagnostics.append("continuity_or_source_unavailable") + # A healthy derived cache is never a real M2 acceptance receipt. + receipt = None + if config.project_m2_receipt: + try: + with config.project_m2_receipt.open("rb") as handle: + raw = handle.read(16385) + if len(raw) <= 16384: + receipt = json.loads(raw) + except (OSError, ValueError): + pass + if (isinstance(receipt, dict) and receipt.get("schema") == "memorymaster.project-m2-probe.v1" + and receipt.get("independent") is True and receipt.get("installed") is True + and receipt.get("passed") == 3 and len(views) == 3 + and receipt.get("views") == views and receipt.get("prohibited_tool_calls") == 0 + and receipt.get("source_revision") and receipt.get("observed_at") + and receipt.get("epochs") == {ctx.scope: ctx.epoch for ctx in contexts}): + counts["m2_passed"] = 3 + else: + diagnostics.append("real_m2_probe_pending") + flags = [name for name in ("REFRESH", "SESSION_START", "COCKPIT") if active(PREFIX + name)] + detail = "enabled=" + ",".join(flags) + "; " + ",".join(dict.fromkeys(diagnostics)) + return ReviewResult("project_projection", Verdict.FAIL if failed else Verdict.WARN if diagnostics else Verdict.PASS, + detail, counts) + + REVIEW_CHECKS = (check_runtime, check_database, check_feature_activation, check_graph_observations, check_compiled_profile, check_recent_private_context, check_retrieval, check_jev_decisions, check_checkpoint_delivery, check_beat_health, check_model_preflight, check_snapshot_dir, check_scheduled_slots, - check_backfill_burndown) + check_project_projection, check_backfill_burndown) def exit_code(results: Iterable[ReviewResult]) -> int: diff --git a/memorymaster/profile/__main__.py b/memorymaster/profile/__main__.py index f6f8a03e..fba4cbeb 100644 --- a/memorymaster/profile/__main__.py +++ b/memorymaster/profile/__main__.py @@ -24,6 +24,20 @@ def _parser() -> argparse.ArgumentParser: run.add_argument("--output-dir", default="") run.add_argument("--force", action="store_true") run.add_argument("--max-map-calls", type=int) + for name in ("project-build", "project-read"): + project = commands.add_parser(name, help="Explicitly build/read a governed project projection") + project.add_argument("--db", required=True) + project.add_argument("--scope", required=True) + project.add_argument("--tenant", required=True, help="tenant key, or legacy-null for NULL only") + project.add_argument("--context-key", required=True) + project.add_argument("--epoch", required=True) + project.add_argument("--project-root", required=True) + project.add_argument("--output-dir", required=True) + project.add_argument("--state-dir", required=True) + project.add_argument("--ledger-dir") + project.add_argument("--ledger-repo") + project.add_argument("--ledger-filtered", action="store_true") + project.add_argument("--alias", action="append", default=[]) return parser @@ -42,7 +56,37 @@ def _status(db_path: Path) -> dict[str, object]: def main(argv: list[str] | None = None) -> int: args = _parser().parse_args(argv) - if args.command == "status": + if args.command.startswith("project-"): + from memorymaster.profile.project_projection import ( + ProjectProjectionConfig, + build_project_projection, + publish_project_projection, + read_project_projection, + ) + from memorymaster.profile.project_runtime import PREFIX, active, resolve_projection_scope + + if not active(PREFIX + "ENABLED"): + result = {"status": "inactive"} + else: + config = ProjectProjectionConfig( + db_path=Path(args.db), scope=args.scope, + tenant_id=None if args.tenant == "legacy-null" else args.tenant, + context_key=args.context_key, epoch=args.epoch, project_root=Path(args.project_root), + output_root=Path(args.output_dir), state_root=Path(args.state_dir), + ledger_root=Path(args.ledger_dir) if args.ledger_dir else None, + ledger_repo=args.ledger_repo, ledger_filtered=args.ledger_filtered, + aliases=tuple(args.alias), enabled=True) + if resolve_projection_scope(config, config.project_root) is None: + result = {"ok": False, "status": "unauthorized_project"} + elif args.command == "project-build": + projection = build_project_projection(config) + result = {"ok": publish_project_projection(config, projection), + "status": projection.status, "manifest": projection.public_manifest} + else: + projection = read_project_projection(config) + result = ({"payload": projection.payload, "manifest": projection.public_manifest, + "status": projection.status} if projection else {"ok": False, "status": "unavailable"}) + elif args.command == "status": result = _status(Path(args.db)) else: from memorymaster.core.service import MemoryService diff --git a/memorymaster/profile/project_projection.py b/memorymaster/profile/project_projection.py new file mode 100644 index 00000000..1a09b1a6 --- /dev/null +++ b/memorymaster/profile/project_projection.py @@ -0,0 +1,649 @@ +"""Deterministic, explicitly authorized project projections and validated views.""" + +from __future__ import annotations + +from contextlib import contextmanager +from dataclasses import dataclass, field +from datetime import datetime, timezone +import hashlib +import json +import os +from pathlib import Path +import re +import tempfile +import time +from typing import Iterator + +from memorymaster.core.generated_artifacts import ( + GENERATED_OPENER as OPENER, + GENERATED_CLOSER as CLOSER, +) +from memorymaster.profile.egress import prepare_claim_egress +from memorymaster.profile.project_repository import ( + _display, + digest, + load_project_snapshot, + utc_now, +) + +SCHEMA = "memorymaster.project-projection.v1" +SECTIONS = ("Purpose", "Constraints", "Decisions", "Open cards", "Gotchas") +MAX_MANIFEST_BYTES = 256 * 1024 + + +@dataclass(frozen=True) +class ProjectProjectionConfig: + db_path: Path + scope: str + tenant_id: str | None + context_key: str + epoch: str + project_root: Path + output_root: Path + state_root: Path + ledger_root: Path | None = None + ledger_repo: str | None = None + ledger_filtered: bool = False + aliases: tuple[str, ...] = () + enabled: bool = False + budget_bytes: int = 2200 + + def __post_init__(self) -> None: + if str(self.db_path).startswith(("postgres:", "postgresql:")): + raise ValueError("SQLite required") + for name in ("context_key", "epoch"): + if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,95}", getattr(self, name)): + raise ValueError("invalid context identity") + if not re.fullmatch(r"project:[a-z0-9][a-z0-9_.-]{0,95}", self.scope): + raise ValueError("invalid project scope") + if ( + any(not re.fullmatch(r"project:[a-z0-9][a-z0-9_.-]{0,95}", alias) for alias in self.aliases) + or len(self.aliases) > 8 + ): + raise ValueError("invalid aliases") + if not 400 <= self.budget_bytes <= 2200: + raise ValueError("project cap must be 400..2200 UTF-8 bytes") + for name in ("db_path", "project_root", "output_root", "state_root", "ledger_root"): + value = getattr(self, name) + if value is not None: + object.__setattr__(self, name, Path(value).resolve()) + object.__setattr__(self, "aliases", tuple(self.aliases)) + + @property + def slug(self) -> str: + return self.scope.split(":", 1)[1] + + +def projection_paths(config: ProjectProjectionConfig) -> tuple[Path, Path]: + root = config.output_root / config.context_key + return root / f"project-{config.slug}.md", root / f"project-{config.slug}.json" + + +@contextmanager +def project_lock(path: Path, timeout: float = 0.05) -> Iterator[bool]: + """Nonblocking kernel lock; the OS releases ownership after process death.""" + path = Path(path) + path.parent.mkdir(parents=True, exist_ok=True) + stream = path.open("a+b") + acquired = False + try: + if stream.tell() == 0: + stream.write(b"\0") + stream.flush() + deadline = time.monotonic() + max(0, timeout) + while True: + try: + stream.seek(0) + if os.name == "nt": + import msvcrt + + msvcrt.locking(stream.fileno(), msvcrt.LK_NBLCK, 1) + else: + import fcntl + + fcntl.flock(stream.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + acquired = True + break + except OSError: + if time.monotonic() >= deadline: + break + time.sleep(0.005) + yield acquired + finally: + if acquired: + stream.seek(0) + if os.name == "nt": + import msvcrt + + msvcrt.locking(stream.fileno(), msvcrt.LK_UNLCK, 1) + else: + import fcntl + + fcntl.flock(stream.fileno(), fcntl.LOCK_UN) + stream.close() + + +def _atomic_write(path: Path, data: bytes) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + fd, temporary = tempfile.mkstemp(prefix="." + path.name + ".", dir=path.parent) + try: + with os.fdopen(fd, "wb") as stream: + stream.write(data) + stream.flush() + os.fsync(stream.fileno()) + os.replace(temporary, path) + finally: + if os.path.exists(temporary): + os.unlink(temporary) + + +def atomic_json(path: Path, value: dict) -> None: + _atomic_write(path, json.dumps(value, sort_keys=True, ensure_ascii=False, separators=(",", ":")).encode()) + + +@dataclass(frozen=True) +class ProjectionItem: + key: str + section: str + text: str + support: dict + sort: list = field(default_factory=list) + + +def _public_identity(value: str) -> str: + """Keep authorization identities private when egress flags their spelling.""" + prepared = prepare_claim_egress(value) + if prepared.blocked or prepared.findings: + return "opaque-" + digest(value)[:16] + return prepared.text + + +@dataclass +class ProjectProjection: + payload: str + manifest: dict + items: list[dict] + status: str + + @property + def markdown(self) -> str: + return f"{OPENER}\n{self.payload}\n{CLOSER}\n" + + @property + def public_manifest(self) -> dict: + # Export an explicit schema. The private manifest retains the exact + # support identities/versions used for rehydration; ledger metadata is + # never recursively copied onto the CLI surface. The opaque key maps + # back to its private record by hashing that record's support identity. + public = {key: self.manifest[key] for key in ( + "schema", "algorithm_version", + "generation_id", "view_id", "read_at", "checked_at", "output_sha256", + "payload_sha256", "output_bytes", "payload_bytes", "budget_bytes", + "omissions", "diagnostics", "status", "complete", "source_complete", + "next_temporal_boundary", + )} + for key in ("scope", "activation_epoch"): + public[key] = _public_identity(self.manifest[key]) + public["items"] = [] + for record in self.manifest["items"]: + supports = [{"kind": support["kind"], + "id": digest([support["kind"], support["id"]]), + "hash": support["hash"]} for support in record["supports"]] + public["items"].append({ + **{key: record[key] for key in ( + "section", "text", "line", "span", "text_hash", "pinned", + )}, + "key": supports[0]["kind"] + ":" + supports[0]["id"], + "supports": supports, + }) + return public + + +def _line(item: dict) -> str: + support = item["support"] + reference = f"claim:{support['id']}" if support["kind"] == "claim" else _display(support["id"], 48) + return f"- {item['text']} [{reference}]" + + +def _fit_card_text(item: dict, byte_budget: int) -> str: + """Keep the current ID/state slot; titles consume only its remaining UTF-8 bytes.""" + support = item["support"] + if not all(isinstance(support.get(key), str) and support[key].strip() for key in ("id", "state")): + raise ValueError("invalid_card_rendering") + prefix = f"{_display(support['id'], 48)} [{_display(support['state'], 24)}]" + available = byte_budget - len(prefix.encode("utf-8")) + if available < 0: + raise ValueError("card_prefix_exceeds_byte_budget") + title = item.get("title") + if not isinstance(title, str): + raise ValueError("invalid_card_rendering") + text = prefix + " " + title + if len(text.encode("utf-8")) <= byte_budget: + return text + if available < 5: + return prefix + retained = title.encode("utf-8")[:available - 4].decode("utf-8", "ignore") + return prefix + " " + retained + "…" + + +def render_project_projection( + config: ProjectProjectionConfig, + items: list[dict], + *, + generation: str, + read_at: str, + coverage: dict | None = None, + diagnostics: list[str] | None = None, + omissions: dict | None = None, + fingerprints: dict | None = None, + source_complete: bool = True, + next_temporal_boundary: str | None = None, +) -> ProjectProjection: + selected = [dict(item) for item in items] + diagnostics = list(diagnostics or []) + omissions = dict(omissions or {}) + decisions = [item for item in selected if item["section"] == "Decisions"] + mandatory_decisions = {item["key"] for item in decisions[:3]} + if len(decisions) < 3: + diagnostics.append("m2_not_ready:decisions_incomplete") + # Purpose and pinned safety prose share a strict byte allocation. Shortening + # literal excerpts keeps references and support identities intact. + safety = [item for item in selected if item["section"] == "Purpose" or item.get("pinned")] + remaining = 400 + # Fund short excerpts first, returning their unused share to longer ones. + # Stable sorting leaves equal-sized items deterministic and display order intact. + for index, item in enumerate(sorted(safety, key=lambda item: len(item["text"].encode("utf-8")))): + cap = remaining // (len(safety) - index) + if cap < 4 and len(item["text"].encode("utf-8")) > cap: + selected.remove(item) + key = "purpose" if item["section"] == "Purpose" else "pinned_constraints" + omissions[key] = omissions.get(key, 0) + 1 + diagnostics.append(f"m2_not_ready:{key}_trimmed") + continue + text = item["text"] + while len(text.encode()) > cap: + text = text[:-1] + if text != item["text"]: + while text and len((text + "…").encode()) > cap: + text = text[:-1] + text += "…" + item["text"] = text + remaining -= len(item["text"].encode()) + diagnostics = list(dict.fromkeys(diagnostics)) + max_bytes = config.budget_bytes + status = "partial" if diagnostics or any(omissions.values()) else "ready" + + def compose() -> tuple[str, str, dict[int, tuple[int, int, int]]]: + nonlocal status + status = "partial" if diagnostics or any(omissions.values()) else "ready" + identity = digest( + [ + generation, + [(item["key"], item["support"], item["text"]) for item in selected], + status, + omissions, + diagnostics, + ] + ) + lines = [f"Project {_public_identity(config.slug)}", f"View {identity[:16]} | {status}"] + coordinates = {} + byte_offset = sum(len(line.encode()) + 1 for line in lines) + + def append(line: str) -> None: + nonlocal byte_offset + lines.append(line) + byte_offset += len(line.encode()) + 1 + + for section in SECTIONS: + append(f"## {section}") + matches = [(index, item) for index, item in enumerate(selected) if item["section"] == section] + if section == "Purpose" and not matches: + append("Purpose not recorded in confirmed memory.") + for index, item in matches: + rendered = _line(item) + # Each selected occurrence owns its coordinates even when + # redaction or truncation gives another item the same text. + coordinates[index] = (len(lines) + 1, byte_offset, byte_offset + len(rendered.encode())) + append(rendered) + if diagnostics: + lines.append("Status: " + ", ".join(dict.fromkeys(diagnostics))) + if any(omissions.values()): + lines.append("Omitted: " + ", ".join(f"{key}={value}" for key, value in sorted(omissions.items()) if value)) + return "\n".join(lines), identity, coordinates + + def size(payload: str) -> int: + return len(f"{OPENER}\n{payload}\n{CLOSER}\n".encode()) + + # Reserve the complete compact card set before optional prose; exact support + # records are generated only after the final trim. + compacted = False + while True: + payload, identity, coordinates = compose() + if ( + size(payload) <= max_bytes + and len(selected) <= 64 + and sum(len(item["support"].get("citations", [])) for item in selected) <= 256 + ): + break + if not compacted: + for item in selected: + text = item["text"] + minimum = 48 if item["section"] == "Open cards" else 80 + if item["section"] == "Open cards": + try: + item["text"] = _fit_card_text(item, minimum) + except ValueError as exc: + if str(exc) != "card_prefix_exceeds_byte_budget": + raise + # Keep its mandatory slots until normal budget eviction + # can remove the card with an explicit omission receipt. + elif len(text) > minimum: + item["text"] = text[: minimum - 1] + "…" + compacted = True + continue + optional = ( + lambda item: item["section"] == "Gotchas", + lambda item: item["section"] == "Decisions" and item["key"] not in mandatory_decisions, + lambda item: item["section"] == "Constraints" and not item.get("pinned"), + lambda item: item["section"] == "Purpose", + lambda item: item["section"] == "Constraints", + lambda item: item["section"] == "Decisions", + lambda item: item["section"] == "Open cards", + ) + removable = next((item for predicate in optional for item in reversed(selected) if predicate(item)), None) + if removable is None: + raise ValueError("mandatory_projection_framing_does_not_fit") + selected.remove(removable) + key = removable["section"].lower().replace(" ", "_") + omissions[key] = omissions.get(key, 0) + 1 + if removable["section"] == "Open cards" and "m2_not_ready:open_cards_trimmed" not in diagnostics: + diagnostics.append("m2_not_ready:open_cards_trimmed") + if removable["section"] == "Decisions" and "m2_not_ready:decisions_trimmed" not in diagnostics: + if removable["key"] in mandatory_decisions: + diagnostics.append("m2_not_ready:decisions_trimmed") + if removable.get("pinned") and "m2_not_ready:pinned_constraints_trimmed" not in diagnostics: + diagnostics.append("m2_not_ready:pinned_constraints_trimmed") + records = [] + for index, item in enumerate(selected): + line, start, end = coordinates[index] + records.append( + { + "key": item["key"], + "section": item["section"], + "text": item["text"], + "line": line, + "span": [start, end], + "text_hash": digest(item["text"]), + "supports": [item["support"]], + "pinned": item.get("pinned", False), + } + ) + markdown = f"{OPENER}\n{payload}\n{CLOSER}\n" + manifest = { + "schema": SCHEMA, + "algorithm_version": 1, + "scope": config.scope, + "aliases": list(config.aliases), + "tenant_id": config.tenant_id, + "context_key": config.context_key, + "activation_epoch": config.epoch, + "generation_id": generation, + "view_id": identity, + "read_at": read_at, + "checked_at": utc_now(), + "output_sha256": hashlib.sha256(markdown.encode()).hexdigest(), + "payload_sha256": hashlib.sha256(payload.encode()).hexdigest(), + "output_bytes": len(markdown.encode()), + "payload_bytes": len(payload.encode()), + "budget_bytes": max_bytes, + "items": records, + "source_coverage": coverage or {}, + "source_fingerprints": fingerprints or {}, + "omissions": omissions, + "diagnostics": list(dict.fromkeys(diagnostics)), + "status": status, + "complete": source_complete, + "source_complete": source_complete, + "next_temporal_boundary": next_temporal_boundary, + } + return ProjectProjection(payload, manifest, selected, status) + + +def build_project_projection(config: ProjectProjectionConfig) -> ProjectProjection: + if not config.enabled: + raise ValueError("projection inactive") + snapshot = load_project_snapshot(config) + ordered = [] + for section in SECTIONS: + matches = [item for item in snapshot["items"].values() if item["section"] == section and item.get("open", True)] + matches.sort(key=lambda item: item["sort"][1:] if section == "Decisions" else item["sort"], + reverse=section == "Decisions") + ordered.extend(matches[:1] if section == "Purpose" else matches[:10] if section == "Decisions" else matches) + diagnostics = list(snapshot["diagnostics"]) + if not snapshot["complete"] and not diagnostics: + diagnostics.append("source_incomplete") + if any(diagnostic.startswith("ledger_") for diagnostic in diagnostics): + diagnostics.append("m2_not_ready:ledger_incomplete") + generation = digest([config.scope, config.context_key, config.epoch, snapshot["read_at"], snapshot["coverage"]]) + return render_project_projection( + config, + ordered, + generation=generation, + read_at=snapshot["read_at"], + coverage=snapshot["coverage"], + diagnostics=diagnostics, + fingerprints=snapshot["fingerprints"], + source_complete=snapshot["complete"], + next_temporal_boundary=snapshot["next_temporal_boundary"], + ) + + +def publish_project_projection(config: ProjectProjectionConfig, projection: ProjectProjection) -> bool: + if not config.enabled: + return False + md, path = projection_paths(config) + with project_lock(path.with_suffix(".lock"), timeout=0) as acquired: + if not acquired: + return False + if len(json.dumps(projection.manifest).encode()) > MAX_MANIFEST_BYTES: + return False + _atomic_write(md, projection.markdown.encode()) + atomic_json(path, projection.manifest) + return True + + +def read_manifest(config: ProjectProjectionConfig) -> dict | None: + if not config.enabled: + return None + md, path = projection_paths(config) + try: + if path.stat().st_size > MAX_MANIFEST_BYTES or md.stat().st_size > config.budget_bytes: + return None + with path.open("rb") as stream: + raw = stream.read(MAX_MANIFEST_BYTES + 1) + with md.open("rb") as stream: + output = stream.read(config.budget_bytes + 1) + manifest = json.loads(raw) + if not isinstance(manifest, dict) or any( + not isinstance(manifest.get(key), expected) + for key, expected in ( + ("omissions", dict), ("source_coverage", dict), ("source_fingerprints", dict), + ("items", list), ("diagnostics", list), ("checked_at", str), + ("next_temporal_boundary", (str, type(None))), + ) + ): + return None + if datetime.fromisoformat(manifest["checked_at"]).tzinfo is None: + return None + # Exported publicly but outside every digest: only an aware ISO instant may pass through. + boundary = manifest["next_temporal_boundary"] + if boundary is not None and datetime.fromisoformat(boundary).tzinfo is None: + return None + if ( + manifest["schema"] != SCHEMA + or manifest["scope"] != config.scope + or manifest["tenant_id"] != config.tenant_id + or manifest["context_key"] != config.context_key + or manifest["activation_epoch"] != config.epoch + or manifest["aliases"] != list(config.aliases) + or manifest["budget_bytes"] != config.budget_bytes + or hashlib.sha256(output).hexdigest() != manifest["output_sha256"] + or len(output) != manifest["output_bytes"] + or len(output) > config.budget_bytes + ): + return None + text = output.decode() + if not text.startswith(OPENER + "\n") or not text.endswith("\n" + CLOSER + "\n"): + return None + payload = text[len(OPENER) + 1 : -len(CLOSER) - 2] + if ( + hashlib.sha256(payload.encode()).hexdigest() != manifest["payload_sha256"] + or len(payload.encode()) != manifest["payload_bytes"] + ): + return None + records = manifest["items"] + if ( + len(records) > 64 + or len(json.dumps(manifest["source_coverage"]).encode()) > 128 * 1024 + or len(manifest["source_coverage"]) > 769 + ): + return None + # A pair hash proves internal consistency, never authority. Only fixed + # status vocabulary may be copied from the disposable cache to output. + allowed_diagnostics = { + "source_incomplete", "ledger_unavailable", "ledger_incomplete", + "claims_unavailable", "coverage_limit", "inactive", + "incremental_scan_fence_incomplete", + *(f"m2_not_ready:{reason}" for reason in ( + "decisions_incomplete", "purpose_trimmed", "pinned_constraints_trimmed", + "open_cards_trimmed", "decisions_trimmed", "ledger_incomplete", + "cards_changed", "context_withdrawn", + )), + } + if (any(not isinstance(value, str) or value not in allowed_diagnostics + for value in manifest["diagnostics"]) + or any(key not in {"purpose", "constraints", "decisions", "open_cards", "gotchas", "pinned_constraints"} + or type(value) is not int or value < 0 + for key, value in manifest["omissions"].items())): + return None + citations = 0 + keys = set() + for record in records: + if (not isinstance(record, dict) or not isinstance(record.get("supports"), list) + or len(record["supports"]) != 1 or not isinstance(record["supports"][0], dict)): + return None + if record["key"] in keys or record["section"] not in SECTIONS: + return None + keys.add(record["key"]) + support = record["supports"][0] + if support["kind"] == "claim": + if (type(support["id"]) is not int or support["id"] <= 0 + or not isinstance(support.get("citations"), list)): + return None + citations += len(support["citations"]) + elif support["kind"] != "card" or not isinstance(support["id"], str): + return None + if record["text_hash"] != digest(record["text"]): + return None + rendered = _line({"text": record["text"], "support": support}) + line_index = record["line"] - 1 + lines = payload.splitlines() + if line_index < 0 or line_index >= len(lines) or lines[line_index] != rendered: + return None + start = len(("\n".join(lines[:line_index]) + "\n").encode()) if line_index else 0 + if record["span"] != [start, start + len(rendered.encode())]: + return None + if citations > 256: + return None + expected_generation = digest( + [config.scope, config.context_key, config.epoch, manifest["read_at"], manifest["source_coverage"]] + ) + expected_view = digest( + [ + manifest["generation_id"], + [(record["key"], record["supports"][0], record["text"]) for record in records], + manifest["status"], + manifest["omissions"], + manifest["diagnostics"], + ] + ) + if ( + manifest["generation_id"] != expected_generation + or manifest["view_id"] != expected_view + or f"View {expected_view[:16]} | {manifest['status']}" not in payload.splitlines() + ): + return None + return manifest + except (OSError, ValueError, KeyError, TypeError, UnicodeError): + return None + + +def read_project_projection(config: ProjectProjectionConfig) -> ProjectProjection | None: + manifest = read_manifest(config) + if manifest is None: + return None + try: + checked = datetime.fromisoformat(manifest["checked_at"]) + if checked.tzinfo is None or (datetime.now(timezone.utc) - checked).total_seconds() > 3600: + return None + except (ValueError, TypeError, KeyError): + return None + ids = [record["supports"][0]["id"] for record in manifest["items"] if record["supports"][0]["kind"] == "claim"] + current = load_project_snapshot(config, claim_ids=ids, validation=True) + selected = [] + diagnostics = list(manifest["diagnostics"]) + omissions = dict(manifest["omissions"]) + card_spare_bytes = max(0, config.budget_bytes - manifest["output_bytes"]) + for record in manifest["items"]: + item = current["items"].get(record["key"]) + if record["supports"][0]["kind"] == "card": + # Modified still-open cards are safe to render from the authorized + # ledger snapshot and retain their reserved place in the payload. + if item and item.get("open"): + reserved = len(record["text"].encode("utf-8")) + unchanged = item["support"] == record["supports"][0] + try: + # A title edit cannot take space reserved for an unrelated card. + fitted = _fit_card_text(item, reserved + (0 if unchanged else card_spare_bytes)) + except ValueError as exc: + if str(exc) != "card_prefix_exceeds_byte_budget": + raise + omissions["open_cards"] = omissions.get("open_cards", 0) + 1 + diagnostics.append("m2_not_ready:open_cards_trimmed") + card_spare_bytes += reserved + continue + item = {**item, "text": fitted} + card_spare_bytes += reserved - len(fitted.encode("utf-8")) + selected.append(item) + elif "m2_not_ready:cards_changed" not in diagnostics: + diagnostics.append("m2_not_ready:cards_changed") + elif item and item["support"] == record["supports"][0]: + # Retain only the cached byte reservation. Every factual character + # comes from the rehydrated, authorized source, including excerpts. + reserved = len(record["text"].encode("utf-8")) + text = item["text"] + if len(text.encode("utf-8")) > reserved: + text = (text.encode("utf-8")[:max(0, reserved - 3)].decode("utf-8", "ignore") + "…" + if reserved >= 3 else "") + selected.append({**item, "text": text}) + else: + diagnostics.append("m2_not_ready:context_withdrawn") + diagnostics.extend(current["diagnostics"]) + if any(key.startswith("card:") and key not in manifest["source_coverage"] for key in current["coverage"]): + diagnostics.append("m2_not_ready:cards_changed") + if not current["complete"] and "source_incomplete" not in diagnostics: + diagnostics.append("source_incomplete") + result = render_project_projection( + config, + selected, + generation=manifest["generation_id"], + read_at=manifest["read_at"], + coverage=manifest["source_coverage"], + diagnostics=list(dict.fromkeys(diagnostics)), + omissions=omissions, + fingerprints=manifest["source_fingerprints"], + source_complete=current["complete"], + next_temporal_boundary=manifest.get("next_temporal_boundary"), + ) + result.manifest["checked_at"] = manifest["checked_at"] + return result diff --git a/memorymaster/profile/project_repository.py b/memorymaster/profile/project_repository.py new file mode 100644 index 00000000..54698119 --- /dev/null +++ b/memorymaster/profile/project_repository.py @@ -0,0 +1,627 @@ +"""Bounded read-only project evidence, never a writable profile repository.""" + +from __future__ import annotations + +from contextlib import closing +from bisect import bisect_left, bisect_right +from datetime import datetime, timezone +import hashlib +import json +from pathlib import Path +import re +import sqlite3 +import time +from types import SimpleNamespace +from typing import Any + +from memorymaster.core.generated_artifacts import contains_generated_marker +from memorymaster.core.models import _parse_iso_strict +from memorymaster.core.security import scan_persisted_value +from memorymaster.core.temporal_policy import claim_is_temporally_current +from memorymaster.profile.egress import prepare_claim_egress + +OPEN_STATES = frozenset({"queued", "ready", "running", "review", "blocked"}) +MAX_CLAIMS = 512 +MAX_CITATIONS = 2048 +MAX_FIELD_BYTES = 16384 +MAX_SOURCE_BYTES = 8 * 1024 * 1024 +GENERATED_SOURCES = frozenset( + {"project-projection", "compiled-profile", "cockpit", "operational-review", "beat-narrative", "graph-summary"} +) + + +def digest(value: Any) -> str: + return hashlib.sha256( + json.dumps(value, sort_keys=True, ensure_ascii=False, separators=(",", ":")).encode() + ).hexdigest() + + +def utc_now() -> str: + return datetime.now(timezone.utc).isoformat() + + +def _instant_key(instant: datetime) -> int: + """Exact UTC microseconds, including instants outside UTC's calendar range.""" + elapsed = instant - datetime.min.replace(tzinfo=timezone.utc) + return (elapsed.days * 86400 + elapsed.seconds) * 1_000_000 + elapsed.microseconds + + +def _boundary_key(entry: list) -> tuple[int, int, str]: + return _instant_key(datetime.fromisoformat(entry[0])), entry[1], entry[0] + + +def _validity_instants(rows) -> list[tuple[int, datetime]]: + """Compare accepted ISO forms as instants without rewriting support fields.""" + result = [] + for row in rows: + for name in ("valid_from", "valid_until"): + try: + instant = _parse_iso_strict(name, row[name]) + except ValueError: + continue # Malformed persisted bounds are ineligible at rehydration. + if instant is not None: + # Normalize display hints only when the UTC calendar can hold + # them. Fence ordering uses integer instants in either case. + if 0 <= _instant_key(instant) <= _instant_key(datetime.max.replace(tzinfo=timezone.utc)): + instant = instant.astimezone(timezone.utc) + result.append((row["id"], instant)) + return result + + +def _safe(value: Any) -> bool: + def marked(raw: Any) -> bool: + if isinstance(raw, str): + return contains_generated_marker(raw) + if isinstance(raw, dict): + return any(marked(key) or marked(item) for key, item in raw.items()) + if isinstance(raw, (list, tuple)): + return any(marked(item) for item in raw) + return False + + text = json.dumps(value, ensure_ascii=False) + lowered = text.lower() + return ( + len(text.encode()) <= MAX_FIELD_BYTES + and not marked(value) + and not any(marker in lowered for marker in ("generated by memorymaster", "[redacted", "[erased_claim_text]")) + and not scan_persisted_value(value) + and not prepare_claim_egress(text).blocked + ) + + +def _display(text: Any, limit: int = 240) -> str: + # One inert line; never permit source headings or markers to create instructions. + text = prepare_claim_egress(str(text or "")).text + text = re.sub(r"[\x00-\x1f\x7f]", " ", text) + text = text.translate(str.maketrans({"<": "‹", ">": "›", "[": "（", "]": "）", "`": "′"})) + literal = " ".join(text.split()) + return literal if len(literal) <= limit else literal[: limit - 1] + "…" + + +def cheap_fingerprints(config: Any) -> dict[str, Any]: + """Filesystem change hints only: callers must also honor temporal boundaries.""" + if not config.enabled: + return {} + paths = [Path(config.db_path), Path(str(config.db_path) + "-wal")] + if config.ledger_root: + paths.extend(_ledger_paths(config)) + result: dict[str, Any] = {} + for index, path in enumerate(paths): + try: + stat = path.stat() + result[str(index)] = [stat.st_mtime_ns, stat.st_size] + except OSError: + result[str(index)] = None + return result + + +def _ledger_paths(config: Any) -> list[Path]: + root = Path(config.ledger_root) + if root.is_file(): + return [root] + # Enumerate only the explicit approved input directory, with a hard count cap. + paths = [] + for path in root.iterdir(): + if path.suffix == ".json" and path.is_file(): + paths.append(path) + if len(paths) > 1024: + raise ValueError("ledger_file_limit") + return sorted(paths) + + +def load_ledger(config: Any, *, max_bytes: int = MAX_SOURCE_BYTES) -> tuple[dict, bool, list[str]]: + if not config.ledger_root or not config.ledger_repo or not config.ledger_filtered: + return {}, False, ["ledger_unavailable"] + consumed = 0 + + def read_bytes(path: Path) -> bytes: + nonlocal consumed + remaining = max_bytes - consumed + if remaining < 0: + raise ValueError("ledger_byte_limit") + with path.open("rb") as stream: + raw = stream.read(remaining + 1) + consumed += len(raw) + if consumed > max_bytes: + raise ValueError("ledger_byte_limit") + return raw + + for attempt in range(2): + try: + paths = _ledger_paths(config) + if not paths: + raise ValueError("no_ledger_input") + before = [(path.stat().st_mtime_ns, path.stat().st_size) for path in paths] + if sum(size for _, size in before) > max_bytes: + raise ValueError("ledger_byte_limit") + records = [] + complete = True + content_hashes = [] + for path in paths: + raw = read_bytes(path) + content_hashes.append(hashlib.sha256(raw).digest()) + data = json.loads(raw) + if isinstance(data, dict) and "tasks" in data: + complete = complete and data.get("complete") is True + tasks = data["tasks"] + else: + tasks = data if isinstance(data, list) else [data] + if not isinstance(tasks, list): + raise ValueError("invalid_ledger") + records.extend(tasks) + if len(records) > 1024: + raise ValueError("ledger_record_limit") + cards = {} + for row in records: + if not isinstance(row, dict): + raise ValueError("invalid_ledger_record") + if row.get("repo") != config.ledger_repo: + continue + if row.get("blind") or row.get("is_blind") or row.get("visibility", "public") != "public": + continue + selected = {name: row.get(name) for name in ("id", "repo", "state", "title", "version")} + if not all( + isinstance(selected[name], str) and selected[name].strip() + for name in ("id", "repo", "state", "title") + ) or not _safe(selected): + complete = False + continue + card_id = selected["id"] + if card_id in cards: + raise ValueError("duplicate_card") + support = { + "kind": "card", + "id": card_id, + "hash": digest(selected), + "version": selected["version"], + "state": selected["state"], + } + cards[card_id] = { + "key": "card:" + card_id, + "section": "Open cards", + "text": f"{_display(card_id, 48)} [{_display(selected['state'], 24)}] {_display(selected['title'], 80)}", + "title": _display(selected["title"], 80), + "support": support, + "open": selected["state"] in OPEN_STATES, + "sort": [selected["state"], card_id], + } + if len(cards) > 256: + raise ValueError("ledger_card_limit") + after_hashes = [hashlib.sha256(read_bytes(path)).digest() for path in paths] + after = [(path.stat().st_mtime_ns, path.stat().st_size) for path in paths] + if before != after or content_hashes != after_hashes or paths != _ledger_paths(config): + if attempt == 0: + continue + raise ValueError("ledger_changed_during_read") + return cards, complete, [] if complete else ["ledger_incomplete"] + except (OSError, ValueError, TypeError, KeyError): + return {}, False, ["ledger_unavailable"] + return {}, False, ["ledger_unavailable"] + + +def _section(row: dict) -> str | None: + mapping = { + "project_purpose": "Purpose", + "purpose": "Purpose", + "constraint": "Constraints", + "rule": "Constraints", + "invariant": "Constraints", + "decision": "Decisions", + "architecture_decision": "Decisions", + "gotcha": "Gotchas", + "pitfall": "Gotchas", + "workaround": "Gotchas", + } + return mapping.get(row.get("predicate")) or mapping.get(row.get("claim_type")) + + +def _claim_item(row: dict, citations: list[dict], *, now: datetime | None = None) -> dict | None: + if ( + row.get("status") != "confirmed" + or (row.get("visibility") or "public") != "public" + or row.get("replaced_by_claim_id") is not None + or row.get("source_agent") in GENERATED_SOURCES + or not citations + or not _section(row) + ): + return None + relevant = { + name: row.get(name) + for name in ( + "text", + "subject", + "predicate", + "object_value", + "claim_type", + "status", + "visibility", + "replaced_by_claim_id", + "source_agent", + "valid_from", + "valid_until", + "scope", + "tenant_id", + "event_time", + "created_at", + "pinned", + ) + } + if not _safe(relevant) or not all(_safe(citation) for citation in citations): + return None + if not claim_is_temporally_current( + SimpleNamespace(valid_from=row.get("valid_from"), valid_until=row.get("valid_until")), now=now + ): + return None + if not str(row.get("text") or "").strip(): + return None + if not all(citation.get("source") and citation.get("locator") for citation in citations): + return None + support = { + "kind": "claim", + "id": int(row["id"]), + "version": int(row.get("version") or 1), + "hash": digest(relevant), + "citations": [{"id": int(c["id"]), "hash": digest(c)} for c in citations], + } + section = _section(row) + date_field = "event_time" if row.get("event_time") else "created_at" + raw_date = row.get(date_field) + date = None + if isinstance(raw_date, str): + try: + date = _parse_iso_strict(date_field, raw_date) + except ValueError: + pass # Keep cited prose, explicitly omit malformed persisted dates. + text = row["text"] + if section == "Decisions" and date is not None: + label = "event" if date_field == "event_time" else "recorded" + text = f"{date.date().isoformat()} ({label}): {text}" + # Egress and inert-line formatting cover the entire composed display, + # including date metadata, rather than only the source prose. + text = _display(text) + # Exact UTC microseconds compare instants without converting to a calendar + # date that can overflow at year 1/9999 with a nonzero offset. Missing dates + # sort before every valid instant (ISO offsets are strictly below 24 hours). + instant_key = -86_400_000_000 + if date is not None: + elapsed = date - datetime.min.replace(tzinfo=timezone.utc) + instant_key = (elapsed.days * 86400 + elapsed.seconds) * 1_000_000 + elapsed.microseconds + return { + "key": f"claim:{row['id']}", + "section": section, + "text": text, + "support": support, + "pinned": bool(row.get("pinned")), + "sort": [0 if row.get("pinned") else 1, instant_key, int(row["id"])], + } + + +def _read_claims( + config: Any, + *, + ids: list[int] | None = None, + changed_since: str | None = None, + validation: bool = False, + max_seen_id: int | None = None, + prior_fence: dict | None = None, + as_of: datetime | None = None, +) -> tuple[dict, bool, dict]: + # One instant governs eligibility and all temporal fence comparisons, + # including when a boundary is crossed while SQLite rows are collected. + as_of = as_of or datetime.fromisoformat(utc_now().replace("Z", "+00:00")) + limit = 64 if validation else MAX_CLAIMS + citation_limit = 256 if validation else MAX_CITATIONS + byte_limit = 2 * 1024 * 1024 if validation else MAX_SOURCE_BYTES + if ids is not None and len(ids) > MAX_CLAIMS: + raise ValueError("claim_limit") + deadline = time.monotonic() + 0.1 + with closing( + sqlite3.connect(Path(config.db_path).resolve().as_uri() + "?mode=ro", uri=True, timeout=0.025) + ) as conn: + conn.row_factory = sqlite3.Row + conn.execute("PRAGMA query_only=ON") + conn.set_progress_handler(lambda: int(time.monotonic() > deadline), 1000) + conn.execute("BEGIN") + try: + clock = conn.execute("SELECT epoch,seq FROM project_source_clock WHERE singleton=1").fetchone() + source_cursor = {"epoch": clock[0], "seq": clock[1]} if clock else None + except sqlite3.OperationalError: + source_cursor = None + scopes = (config.scope, *config.aliases) + params: list[Any] = [config.tenant_id, *scopes] + where = "tenant_id IS ? AND scope IN (" + ",".join("?" for _ in scopes) + ")" + changed_ids = [] + temporal_boundaries = [] + if changed_since: + cursor = (prior_fence or {}).get("source_cursor") + if (not source_cursor or not isinstance(cursor, dict) + or cursor.get("epoch") != source_cursor["epoch"] + or type(cursor.get("seq")) is not int + or not 0 <= cursor["seq"] <= source_cursor["seq"]): + raise ValueError("incremental_cursor_unavailable") + # The transactional cursor covers every insert as well as late + # commits, deletes, scope moves and citation edits. Do not also scan + # IDs above the project's max_seen_id: unchanged unrelated projects + # can have higher IDs and would be traversed again on every start. + instant = datetime.fromisoformat(changed_since.replace("Z", "+00:00")) + if instant.tzinfo is None: + raise ValueError("unqualified_baseline_time") + if instant > as_of: + raise ValueError("future_baseline_time") + watched = (prior_fence or {}).get("watched_ids", []) + if len(watched) > MAX_CLAIMS or any(type(id) is not int or id <= 0 for id in watched): + raise ValueError("invalid_scan_fence") + watched_set = set(watched) + watched_sql = ",".join("?" for _ in watched) or "NULL" + # Protected continuity metadata holds future boundaries ordered by + # integer instants, including bounds beyond the UTC calendar. + # Never reload every watched row to compare ISO forms. + # Older fences lack this contract and must fail closed for rebuild. + temporal_boundaries = (prior_fence or {}).get("temporal_boundaries") + if ((prior_fence or {}).get("temporal_version") != 2 + or not isinstance(temporal_boundaries, list) + or len(temporal_boundaries) > 2 * MAX_CLAIMS + or any(not isinstance(entry, list) or len(entry) != 2 + or not isinstance(entry[0], str) + or type(entry[1]) is not int or entry[1] not in watched_set + for entry in temporal_boundaries) + or temporal_boundaries != sorted(temporal_boundaries, key=_boundary_key)): + raise ValueError("incremental_temporal_fence_unavailable") + lower = bisect_left(temporal_boundaries, _instant_key(instant), + key=lambda entry: _boundary_key(entry)[0]) + upper = bisect_right(temporal_boundaries, _instant_key(as_of), + key=lambda entry: _boundary_key(entry)[0]) + temporal_ids = sorted({id for _, id in temporal_boundaries[lower:upper]}) + temporal_boundaries = temporal_boundaries[upper:] + queries = [ + "SELECT changes.claim_id AS id FROM project_source_changes changes " + "INDEXED BY idx_project_source_changes_seq " + "LEFT JOIN claims ON claims.id=changes.claim_id " + f"WHERE changes.seq > ? AND changes.seq <= ? AND (({where}) " + f"OR changes.claim_id IN ({watched_sql}))", + ] + changes_params: list[Any] = [cursor["seq"], source_cursor["seq"], *params, *watched] + if temporal_ids: + queries.append("SELECT id FROM claims NOT INDEXED WHERE id IN (" + + ",".join("?" for _ in temporal_ids) + ")") + changes_params.extend(temporal_ids) + changed_ids = [row[0] for row in conn.execute( + "SELECT id FROM (" + " UNION ".join(queries) + ") ORDER BY id LIMIT ?", + [*changes_params, limit + 1], + )] + if len(changed_ids) > limit: + return {}, False, {} + ids = sorted(set(ids or ()) | set(changed_ids)) + if ids is not None: + # An empty IN set reads neither claim payloads nor citations. + where += " AND id IN (" + (",".join("?" for _ in ids) or "NULL") + ")" + params.extend(ids) + # Builder scans watch bounded all-status IDs. Ineligible rows can become + # eligible without a reliable post-commit timestamp; their IDs are only + # protected continuity metadata, never factual support or public output. + fields = [row[1] for row in conn.execute("PRAGMA table_info(claims)")] + names = [ + name + for name in fields + if name + in { + "id", + "version", + "text", + "subject", + "predicate", + "object_value", + "claim_type", + "status", + "visibility", + "replaced_by_claim_id", + "source_agent", + "valid_from", + "valid_until", + "scope", + "tenant_id", + "event_time", + "created_at", + "updated_at", + "pinned", + } + ] + # Reject oversized rows before materializing their payload in Python. + size = "+".join(f"length(CAST(COALESCE({name},'') AS BLOB))" for name in names) + columns = ",".join( + f"CASE WHEN ({size}) <= {MAX_FIELD_BYTES} THEN {name} ELSE NULL END AS {name}" if name != "id" else name + for name in names + ) + claim_table = "claims NOT INDEXED" if ids is not None else "claims" + cursor = conn.execute( + f"SELECT {columns}, ({size}) AS _source_bytes FROM {claim_table} WHERE {where} ORDER BY id LIMIT ?", + [*params, limit + 1], + ) + rows = [] + total = 0 + for row in cursor: + if len(rows) >= limit: + fingerprint = {"source_bytes": total} + if not validation: + watched = sorted(set((prior_fence or {}).get("watched_ids", ())) + | set(ids or ()) | {int(value["id"]) for value in rows}) + fingerprint["scan_fence"] = { + "kind": "scan_fence", + "watched_ids": watched[:MAX_CLAIMS], + "max_seen_id": max([max_seen_id or 0, *watched]), + "complete": False, + } + return {}, False, fingerprint + row_bytes = len(json.dumps(dict(row)).encode()) + if total + row_bytes > byte_limit: + # Exhaust the shared source allowance so ledger reads cannot + # restart that budget after an interrupted SQLite stream. + return {}, False, {"source_bytes": byte_limit} + total += row_bytes + rows.append(row) + complete = True + oversized = {row["id"] for row in rows if row["_source_bytes"] > MAX_FIELD_BYTES} + complete = complete and not oversized + ids_read = [row["id"] for row in rows] + citations: dict[int, list] = {} + if ids_read: + placeholders = ",".join("?" for _ in ids_read) + citation_index = " INDEXED BY idx_project_citations_claim" if source_cursor else "" + citation_size = "length(CAST(COALESCE(source,'')||COALESCE(locator,'')||COALESCE(excerpt,'') AS BLOB))" + citation_cursor = conn.execute( + "SELECT id,claim_id," + + ",".join( + f"CASE WHEN {citation_size} <= {MAX_FIELD_BYTES} THEN {name} ELSE NULL END AS {name}" + for name in ("source", "locator", "excerpt") + ) + + f",{citation_size} AS _source_bytes " + f"FROM citations{citation_index} WHERE claim_id IN ({placeholders}) ORDER BY id LIMIT ?", + [*ids_read, citation_limit + 1], + ) + for index, citation in enumerate(citation_cursor): + citation_bytes = len(json.dumps(dict(citation)).encode()) + if total + citation_bytes > byte_limit: + return {}, False, {"source_bytes": byte_limit} + total += citation_bytes + if index >= citation_limit: + return {}, False, {"source_bytes": total} + if citation["_source_bytes"] > MAX_FIELD_BYTES: + oversized.add(citation["claim_id"]) + complete = False + value = dict(citation) + value.pop("_source_bytes") + citations.setdefault(citation["claim_id"], []).append(value) + items = {} + for row in rows: + if row["id"] in oversized: + continue + item = _claim_item(dict(row), citations.get(row["id"], []), now=as_of) + if item: + items[item["key"]] = item + # Refresh normalized metadata only for source rows materialized because + # they changed, crossed a boundary or were explicitly requested. + refreshed_ids = set(changed_ids) | set(ids_read) + temporal_boundaries = sorted([ + *[entry for entry in temporal_boundaries if entry[1] not in refreshed_ids], + *[[boundary.isoformat(timespec="microseconds"), id] for id, boundary in _validity_instants(rows) + if _instant_key(boundary) > _instant_key(as_of)], + ], key=_boundary_key) + fingerprint = { + "data_version": conn.execute("PRAGMA data_version").fetchone()[0], + "next_temporal_boundary": (datetime.fromisoformat(temporal_boundaries[0][0]).isoformat() + if temporal_boundaries else None), + "source_bytes": total, + "changed_ids": changed_ids, + } + if not validation: + watched_ids = sorted(set((prior_fence or {}).get("watched_ids", ())) + | set(ids or ()) | {int(row["id"]) for row in rows}) + fingerprint["scan_fence"] = { + "kind": "scan_fence", + "watched_ids": watched_ids[:MAX_CLAIMS], + "max_seen_id": max([max_seen_id or 0, *watched_ids]), + "complete": complete and len(watched_ids) <= MAX_CLAIMS and source_cursor is not None, + "source_cursor": source_cursor, + "temporal_version": 2, + "temporal_boundaries": temporal_boundaries, + } + complete = complete and len(watched_ids) <= MAX_CLAIMS + return items, complete, fingerprint + + +def snapshot_since(config: Any, coverage: dict, read_at: str) -> dict: + return load_project_snapshot(config, coverage=coverage, changed_since=read_at) + + +def load_project_snapshot( + config: Any, + *, + coverage: dict | None = None, + changed_since: str | None = None, + claim_ids: list[int] | None = None, + validation: bool = False, +) -> dict: + result: dict[str, Any] = { + "read_at": utc_now(), + "coverage": {}, + "items": {}, + "complete": False, + "diagnostics": [], + "fingerprints": {}, + "next_temporal_boundary": None, + } + if not config.enabled: + result["diagnostics"] = ["inactive"] + return result + try: + ids = claim_ids + prior_fence = None + if coverage is not None: + if not changed_since: + raise ValueError("incremental_read_time_unavailable") + prior_fence = coverage.get("__scan_fence__") + # Coverage already holds unchanged support hashes. Only explicit + # display supports and indexed changes need source materialization. + ids = claim_ids or [] + items, complete, fingerprints = _read_claims( + config, + ids=ids, + changed_since=changed_since, + validation=validation, + max_seen_id=prior_fence.get("max_seen_id") if isinstance(prior_fence, dict) else None, + prior_fence=prior_fence, + as_of=datetime.fromisoformat(result["read_at"].replace("Z", "+00:00")), + ) + if coverage is not None and (not prior_fence or not prior_fence.get("complete")): + complete = False + result["diagnostics"].append("incremental_scan_fence_incomplete") + except (sqlite3.Error, OSError, ValueError, TypeError): + items, complete, fingerprints = {}, False, {} + result["diagnostics"].append("claims_unavailable") + byte_budget = 2 * 1024 * 1024 if validation else MAX_SOURCE_BYTES + cards, ledger_complete, diagnostics = load_ledger( + config, max_bytes=max(0, byte_budget - int(fingerprints.get("source_bytes", 0))) + ) + items.update((card["key"], card) for card in cards.values()) + result.update(items=items, complete=complete and ledger_complete, fingerprints=fingerprints) + result["next_temporal_boundary"] = fingerprints.get("next_temporal_boundary") + carried = {key: value for key, value in (coverage or {}).items() if key.startswith("claim:")} + for id in fingerprints.pop("changed_ids", []): + carried.pop(f"claim:{id}", None) + result["coverage"] = {**carried, **{ + key: item["support"] for key, item in items.items() if item.get("open", True) + }} + fence = fingerprints.pop("scan_fence", None) + if fence is not None: + fence["complete"] = fence["complete"] and complete + result["coverage"]["__scan_fence__"] = fence + result["diagnostics"].extend(diagnostics) + if len(json.dumps(result["coverage"]).encode()) > 128 * 1024: + result.update(coverage={}, complete=False) + result["diagnostics"].append("coverage_limit") + return result + + +def load_confirmed_project_claims(config: Any) -> list[dict]: + snapshot = load_project_snapshot(config) + return [item for key, item in snapshot["items"].items() if key.startswith("claim:")] diff --git a/memorymaster/profile/project_runtime.py b/memorymaster/profile/project_runtime.py new file mode 100644 index 00000000..99c208d0 --- /dev/null +++ b/memorymaster/profile/project_runtime.py @@ -0,0 +1,310 @@ +"""Explicit opt-in configuration and bounded project projection consumers.""" + +from __future__ import annotations + +import json +import hashlib +import os +import sqlite3 +import time +from itertools import islice +from datetime import datetime, timezone +from pathlib import Path + +from memorymaster.core.generated_artifacts import GENERATED_CLOSER, GENERATED_OPENER + +PREFIX = "MEMORYMASTER_PROJECT_PROJECTION_" +_refresh_cache: dict[tuple, tuple] = {} +_connections: dict[str, sqlite3.Connection] = {} +_rotation = 0 + + +def active(name: str) -> bool: + return os.environ.get(name, "").strip().lower() in {"1", "true", "yes", "on"} + + +def load_project_configs() -> list: + """Parse an explicit pilot allowlist. No path discovery or store initialization.""" + if not active(PREFIX + "ENABLED"): + return [] + from memorymaster.profile.project_projection import ProjectProjectionConfig + + try: + raw = os.environ.get(PREFIX + "PROJECTS", "[]") + if len(raw.encode()) > 32768: + return [] + mappings = json.loads(raw) + epoch = os.environ.get(PREFIX + "EPOCH", "").strip() + output = os.environ.get(PREFIX + "DIR", "") + state = os.environ.get("MEMORYMASTER_PROJECT_SESSION_STATE_DIR", "") + ledger = os.environ.get("MEMORYMASTER_PROJECT_LEDGER_TASKS_DIR", "") + if (not isinstance(mappings, list) or not 1 <= len(mappings) <= 3 or not epoch + or len(epoch) > 128 or not output or not state + or not Path(output).is_absolute() or not Path(state).is_absolute() + or (ledger and not Path(ledger).is_absolute())): + return [] + configs = [] + seen = set() + for row in mappings: + # Null tenant is a deliberate legacy boundary, not a default across tenants. + if "tenant_id" not in row or not Path(row["db_path"]).is_absolute(): + return [] + if not Path(row["project_root"]).is_absolute(): + return [] + identity = (row["context_key"], row["scope"]) + if identity in seen: + return [] + seen.add(identity) + configs.append(ProjectProjectionConfig( + db_path=Path(row["db_path"]), scope=row["scope"], tenant_id=row["tenant_id"], + context_key=row["context_key"], epoch=epoch, project_root=Path(row["project_root"]), + output_root=Path(output), state_root=Path(state), + ledger_root=Path(ledger) if ledger else None, ledger_repo=row.get("ledger_repo"), + ledger_filtered=row.get("ledger_filtered") is True, + aliases=tuple(row.get("aliases", ())), enabled=True)) + return configs + except (ValueError, KeyError, TypeError): + return [] + + +def resolve_projection_scope(config, cwd) -> str | None: + """Both scope resolvers and the actual repository owner must agree.""" + from memorymaster.core.scope_utils import ( + _repository_owner, + resolve_transcript_workspace_scope, + scope_from_cwd_or_user, + ) + + try: + path = Path(cwd).resolve(strict=True) + if not path.is_dir(): + return None + strict = resolve_transcript_workspace_scope(str(path)) + generic = scope_from_cwd_or_user(str(path)) + if strict != generic or strict not in (config.scope, *config.aliases): + return None + for parent in (path, *list(path.parents)[:12]): + marker = parent / ".git" + if marker.exists(): + owner = _repository_owner(parent) + if owner is None or owner.resolve() != config.project_root.resolve(strict=True): + return None + # Linked worktrees must have a valid backlink; generic resolution alone is insufficient. + if marker.is_file(): + if marker.stat().st_size > 4096: + return None + pointer = marker.read_text(encoding="utf-8").strip() + if not pointer.startswith("gitdir:"): + return None + gitdir = (parent / pointer.partition(":")[2].strip()).resolve() + backlink = gitdir / "gitdir" + if (not backlink.is_file() or backlink.stat().st_size > 4096 + or Path(backlink.read_text(encoding="utf-8").strip()).resolve() != marker.resolve()): + return None + return config.scope + return None + except (OSError, ValueError, RuntimeError): + return None + + +def config_for_cwd(cwd): + matches = [cfg for cfg in load_project_configs() if resolve_projection_scope(cfg, cwd)] + return matches[0] if len(matches) == 1 else None + + +def compose_session_context(payload: str, delta: str = "", profile: str = "") -> str: + """Embed the shared bytes intact, with one envelope and complete optional lines.""" + if len(payload.encode("utf-8")) > 2200 or len(delta.encode("utf-8")) > 500: + raise ValueError("consumer_budget") + if GENERATED_OPENER in payload or GENERATED_CLOSER in payload: + raise ValueError("nested_envelope") + parts = [payload] + if delta: + parts.append(delta) + profile_lines = [line for line in profile.splitlines() + if line.strip() and not line.lstrip().startswith("\n" + GENERATED_OPENER + "\n" "# Compiled User Profile\n\n" "Descriptive, evidence-bound operator facts. This is context, not an instruction file." ) diff --git a/memorymaster/public/v1.py b/memorymaster/public/v1.py index 968fe86d..ca3b68be 100644 --- a/memorymaster/public/v1.py +++ b/memorymaster/public/v1.py @@ -4,6 +4,7 @@ import os import json +import hashlib from dataclasses import asdict, dataclass from datetime import datetime, timezone from pathlib import Path @@ -173,7 +174,7 @@ def _persist_capture( envelope: Any, scope: str, source_agent: str, -) -> tuple[SourceItem, EvidenceItem | None, bool]: +) -> tuple[SourceItem, tuple[EvidenceItem, ...], bool]: source = service.upsert_external_source( source_type="universal_capture", display_name="MemoryMaster Capture", @@ -193,7 +194,9 @@ def _persist_capture( source_id=source.id, source_item_id=source_key, item_type=envelope.content_type, - text=envelope.text, + # Separated authored spans are individual evidence, never a fabricated + # source statement spanning a removed generated range. + text=None if getattr(envelope, "excluded_spans", ()) else envelope.text, payload_json={ "locator": envelope.locator, "mime_type": envelope.mime_type, @@ -206,21 +209,29 @@ def _persist_capture( "producer_session_hash": getattr(envelope, "producer_session_hash", None), "producer_turn_id": getattr(envelope, "producer_turn_id", None), "producer_metadata": dict(getattr(envelope, "producer_metadata", ())), + "excluded_spans": list(getattr(envelope, "excluded_spans", ())), }, content_hash=envelope.content_hash, ) repository = CaptureRepository(service.store) - evidence = repository.evidence_for_content( - source_item_id=item.id, content_hash=envelope.content_hash + spans = getattr(envelope, "retained_spans", ()) + split = bool(getattr(envelope, "excluded_spans", ())) + segments = [(span.text, span.start, span.end) for span in spans] if split else ( + [(envelope.text, 0, len(envelope.text))] if envelope.text is not None else [] ) - deduplicated = evidence is not None or ( - existing_source is not None and envelope.text is None - ) - if evidence is None and envelope.text is not None: + evidences: list[EvidenceItem] = [] + deduplicated = existing_source is not None + for text, start, end in segments: + evidence_hash = hashlib.sha256(f"{envelope.content_hash}:{start}:{end}".encode()).hexdigest() if split else envelope.content_hash + evidence = repository.evidence_for_content(source_item_id=item.id, content_hash=evidence_hash) + deduplicated &= evidence is not None + if evidence is not None: + evidences.append(evidence) + continue evidence = service.add_evidence_item( source_item_id=item.id, evidence_type=envelope.evidence_type or "text", - text=envelope.text, + text=text, provider="memorymaster-capture", confidence=1.0, payload_json={ @@ -231,10 +242,14 @@ def _persist_capture( "producer_session_hash": getattr(envelope, "producer_session_hash", None), "producer_turn_id": getattr(envelope, "producer_turn_id", None), "producer_metadata": dict(getattr(envelope, "producer_metadata", ())), + "original_content_hash": envelope.content_hash, + "source_start": start, + "source_end": end, }, - content_hash=envelope.content_hash, + content_hash=evidence_hash, ) - return item, evidence, deduplicated + evidences.append(evidence) + return item, tuple(evidences), deduplicated def _queue_capture_jobs( @@ -310,19 +325,25 @@ def remember( source_agent=source_agent, platform=platform, ) - item, evidence, evidence_duplicate = _persist_capture( + item, evidences, evidence_duplicate = _persist_capture( service, envelope=envelope, scope=resolved.scope, source_agent=source_agent, ) - jobs, jobs_duplicate = _queue_capture_jobs( - CaptureRepository(service.store), - source_item_id=item.id, - content_hash=envelope.content_hash, - has_evidence=evidence is not None, - blocked_code=envelope.blocked_code, - ) + jobs: tuple[int, ...] = () + jobs_duplicate = True + for evidence in evidences or (None,): + queued, duplicate = _queue_capture_jobs( + CaptureRepository(service.store), + source_item_id=item.id, + content_hash=evidence.content_hash if evidence is not None else envelope.content_hash, + has_evidence=evidence is not None, + blocked_code=envelope.blocked_code, + ) + jobs += queued + jobs_duplicate &= duplicate + evidence = evidences[0] if evidences else None warnings = tuple( dict.fromkeys( (*envelope.warning_codes, *((envelope.blocked_code,) if envelope.blocked_code else ())) diff --git a/memorymaster/stores/migrations/0029_project_source_cursor.py b/memorymaster/stores/migrations/0029_project_source_cursor.py new file mode 100644 index 00000000..434a0e23 --- /dev/null +++ b/memorymaster/stores/migrations/0029_project_source_cursor.py @@ -0,0 +1,115 @@ +"""Track per-claim project-source changes without reading source payloads. + +The project projection uses this cursor to read only the claims whose source +rows changed since its previous SessionStart. Timestamps are not a safe +watermark: a transaction may commit after a read while retaining an older +``updated_at`` value. The clock therefore advances inside the source-row +transaction and leaves a latest-sequence tombstone for deleted claims. +""" + +from __future__ import annotations + +from typing import Any + + +VERSION = 29 +DESCRIPTION = "SQLite project-source per-claim change cursor" +# This adds only derived metadata and triggers. Claim and citation data remain +# valid across a v28/v29 bridge exchange. +MERGE_COMPATIBLE = True + + +_SQLITE = """ +CREATE TABLE IF NOT EXISTS project_source_clock ( + singleton INTEGER PRIMARY KEY CHECK (singleton = 1), + epoch TEXT NOT NULL, + seq INTEGER NOT NULL +); +CREATE TABLE IF NOT EXISTS project_source_changes ( + claim_id INTEGER PRIMARY KEY, + seq INTEGER NOT NULL +); +CREATE INDEX IF NOT EXISTS idx_project_source_changes_seq + ON project_source_changes(seq, claim_id); + +INSERT OR IGNORE INTO project_source_clock(singleton, epoch, seq) +VALUES (1, lower(hex(randomblob(16))), 0); + +CREATE INDEX IF NOT EXISTS idx_project_claims_updated + ON claims(tenant_id, scope, updated_at, id); +CREATE INDEX IF NOT EXISTS idx_project_claims_valid_from + ON claims(tenant_id, scope, valid_from, id); +CREATE INDEX IF NOT EXISTS idx_project_claims_valid_until + ON claims(tenant_id, scope, valid_until, id); +CREATE INDEX IF NOT EXISTS idx_project_citations_created + ON citations(created_at, claim_id); +CREATE INDEX IF NOT EXISTS idx_project_citations_claim + ON citations(claim_id, id); + +CREATE TRIGGER IF NOT EXISTS project_source_claims_ai AFTER INSERT ON claims BEGIN + UPDATE project_source_clock SET seq = seq + 1 WHERE singleton = 1; + INSERT INTO project_source_changes(claim_id, seq) + SELECT NEW.id, seq FROM project_source_clock WHERE singleton = 1 + ON CONFLICT(claim_id) DO UPDATE SET seq = excluded.seq; +END; +CREATE TRIGGER IF NOT EXISTS project_source_claims_au AFTER UPDATE ON claims BEGIN + UPDATE project_source_clock SET seq = seq + 1 WHERE singleton = 1; + INSERT INTO project_source_changes(claim_id, seq) + SELECT OLD.id, seq FROM project_source_clock WHERE singleton = 1 + ON CONFLICT(claim_id) DO UPDATE SET seq = excluded.seq; + INSERT INTO project_source_changes(claim_id, seq) + SELECT NEW.id, seq FROM project_source_clock WHERE singleton = 1 + ON CONFLICT(claim_id) DO UPDATE SET seq = excluded.seq; +END; +CREATE TRIGGER IF NOT EXISTS project_source_claims_ad AFTER DELETE ON claims BEGIN + UPDATE project_source_clock SET seq = seq + 1 WHERE singleton = 1; + INSERT INTO project_source_changes(claim_id, seq) + SELECT OLD.id, seq FROM project_source_clock WHERE singleton = 1 + ON CONFLICT(claim_id) DO UPDATE SET seq = excluded.seq; +END; + +CREATE TRIGGER IF NOT EXISTS project_source_citations_ai AFTER INSERT ON citations BEGIN + UPDATE project_source_clock SET seq = seq + 1 WHERE singleton = 1; + INSERT INTO project_source_changes(claim_id, seq) + SELECT NEW.claim_id, seq FROM project_source_clock WHERE singleton = 1 + ON CONFLICT(claim_id) DO UPDATE SET seq = excluded.seq; +END; +CREATE TRIGGER IF NOT EXISTS project_source_citations_au AFTER UPDATE ON citations BEGIN + UPDATE project_source_clock SET seq = seq + 1 WHERE singleton = 1; + INSERT INTO project_source_changes(claim_id, seq) + SELECT OLD.claim_id, seq FROM project_source_clock WHERE singleton = 1 + ON CONFLICT(claim_id) DO UPDATE SET seq = excluded.seq; + INSERT INTO project_source_changes(claim_id, seq) + SELECT NEW.claim_id, seq FROM project_source_clock WHERE singleton = 1 + ON CONFLICT(claim_id) DO UPDATE SET seq = excluded.seq; +END; +CREATE TRIGGER IF NOT EXISTS project_source_citations_ad AFTER DELETE ON citations BEGIN + UPDATE project_source_clock SET seq = seq + 1 WHERE singleton = 1; + INSERT INTO project_source_changes(claim_id, seq) + SELECT OLD.claim_id, seq FROM project_source_clock WHERE singleton = 1 + ON CONFLICT(claim_id) DO UPDATE SET seq = excluded.seq; +END; +""" + + +def _has_table(conn: Any, name: str) -> bool: + return conn.execute( + "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?", (name,) + ).fetchone() is not None + + +def apply_sqlite(conn: Any) -> None: + """Install the cursor only when both authoritative source tables exist.""" + # The migration framework can stamp an empty database before schema.sql + # creates its tables. Creating half of this cursor would be misleading and + # cannot safely track citation writes, so leave such databases untouched. + if not _has_table(conn, "claims") or not _has_table(conn, "citations"): + return + conn.executescript(_SQLITE) + conn.commit() + + +def apply_postgres(conn: Any) -> None: + """Fail closed until a PostgreSQL source-cursor contract exists.""" + del conn + raise RuntimeError("migration 29 is SQLite-only; project source cursor is SQLite-only") diff --git a/memorymaster/stores/storage.py b/memorymaster/stores/storage.py index 28e44e20..8c6b92fd 100644 --- a/memorymaster/stores/storage.py +++ b/memorymaster/stores/storage.py @@ -215,6 +215,14 @@ def init_db(self) -> None: # A standalone `migrate` may have stamped v17 on an empty database. # Converge again after all prerequisite tables now exist. self._ensure_governed_capture_schema(mig_conn) + from importlib import import_module + + # Migration 29 intentionally no-ops on a bare database. A later + # init must converge its SQLite-only source cursor after schema.sql + # has created claims and citations, even if the runner stamped it. + import_module( + "memorymaster.stores.migrations.0029_project_source_cursor" + ).apply_sqlite(mig_conn) _refresh_planner_statistics(mig_conn) if fastpath: # Stamp ONLY after the full path succeeded end-to-end — an diff --git a/memorymaster/surfaces/scheduled_task.py b/memorymaster/surfaces/scheduled_task.py index b9749ba0..69dbc0a9 100644 --- a/memorymaster/surfaces/scheduled_task.py +++ b/memorymaster/surfaces/scheduled_task.py @@ -153,6 +153,24 @@ def _write_beat_artifact(workspace: str, state_db: Path, guard_log: Path) -> Pat def _run_beat(args: argparse.Namespace) -> int: + """Run capture and the separately gated deterministic phase on every enabled tick.""" + if not _beat_enabled(): + return _run_capture_beat(args) + outcomes: list[str] = [] + try: + return _run_capture_beat(args, outcomes) + finally: + from memorymaster.profile.project_runtime import refresh_project_projections + + try: + result = refresh_project_projections(skip=bool(outcomes and outcomes[0] == "skip")) + if result.get("status") != "inactive": + log_event(_JOB_LOG, "project_projection_tick", surface="job", job="beat", result=result) + except Exception as exc: + log_event(_JOB_LOG, "project_projection_error", surface="job", job="beat", error=type(exc).__name__) + + +def _run_capture_beat(args: argparse.Namespace, outcomes: list[str] | None = None) -> int: """One AO-2 beat tick: scan transcripts, ask the latency guard, then maybe run a bounded dream. Off unless MEMORYMASTER_BEAT_ENABLED is truthy. The guard is asked first; on ``skip`` nothing @@ -177,6 +195,8 @@ def _run_beat(args: argparse.Namespace) -> int: # The guard comes first: a hard breach (skip) must not scan directories, probe files or write # the state db, only record the decision. decision = latency_guard.evaluate() + if outcomes is not None: + outcomes.append(decision.action) scan: dict | None = None dream: dict | None = None if decision.action != latency_guard.SKIP: diff --git a/memorymaster/surfaces/session_end_ingest.py b/memorymaster/surfaces/session_end_ingest.py index 7e351c26..53e978bc 100644 --- a/memorymaster/surfaces/session_end_ingest.py +++ b/memorymaster/surfaces/session_end_ingest.py @@ -29,34 +29,46 @@ def _sensitive(claim: dict[str, Any]) -> bool: return bool(redact_text(joined)[1]) -def _assistant_text(transcript_path: str) -> str: +def _assistant_text(transcript_path: str, *, start_offset: int = 0, end_offset: int | None = None) -> str: + from memorymaster.dreaming.capture import _conversation_record + path = Path(transcript_path) if not path.is_file(): return "" messages: list[str] = [] - lines = path.read_text(encoding="utf-8", errors="replace").splitlines() - for line in reversed(lines[-200:]): + with path.open("rb") as handle: + lines = handle.read(end_offset if end_offset is not None else -1).splitlines(keepends=True) + state: dict[str, Any] = {} + eligible: list[tuple[int, str]] = [] + offset = 0 + for index, line in enumerate(lines): + in_increment = offset >= start_offset + offset += len(line) try: - entry = json.loads(line) + entry = json.loads(line.decode("utf-8", errors="replace")) except json.JSONDecodeError: continue if not isinstance(entry, dict): continue - record = entry - for key in ("message", "payload"): - if isinstance(entry.get(key), dict): - record = entry[key] - break - if str(record.get("role") or "") not in {"assistant", "model"}: - continue - content = record.get("content", "") - if isinstance(content, list): - content = " ".join( - str(part.get("text", "")) - for part in content - if isinstance(part, dict) and part.get("type") in (None, "text") - ) - if isinstance(content, str) and len(content.strip()) > 30: + provider = "codex" if entry.get("type") == "response_item" else "claude" + if provider == "claude": + payload = entry.get("message", entry.get("payload", entry)) + if isinstance(payload, dict): + payload = dict(payload) + role = str(payload.get("role") or "") + if role == "model": + payload["role"] = "assistant" + role = "assistant" + entry = {**entry, "message": payload, "type": role or entry.get("type", "")} + record = _conversation_record(entry, provider, state=state) + # Replay old records for provenance only. Extract solely from the new + # committed byte interval; do not re-distill earlier authored content. + if in_increment and record is not None and record[0] == "assistant": + eligible.extend((index, span.text) for span in record[1]) + for index, content in reversed(eligible): + if index < len(lines) - 200: + break + if len(content.strip()) > 30: messages.append(content[:500]) if sum(map(len, messages)) > 3000: break @@ -120,8 +132,10 @@ def run( *, source_agent: str, cwd: str | None, + start_offset: int = 0, + end_offset: int | None = None, ) -> int: - text = _assistant_text(transcript_path) + text = _assistant_text(transcript_path, start_offset=start_offset, end_offset=end_offset) return ingest_learnings( db_path, _distill(text), source_agent=source_agent, cwd=cwd ) if text else 0 @@ -133,7 +147,15 @@ def main(argv: list[str] | None = None) -> int: parser.add_argument("--transcript", required=True) parser.add_argument("--source-agent", default="codex-session") parser.add_argument("--cwd", default=None) + parser.add_argument("--session-id", help="explicit project lifecycle identity; requires --end-at") + parser.add_argument("--end-at", help="observed UTC end time; requires --session-id") args = parser.parse_args(argv) + if bool(args.session_id) != bool(args.end_at): + parser.error("--session-id and --end-at must be supplied together") + if args.session_id: + from memorymaster.profile.project_runtime import record_session_end + + record_session_end({"cwd": args.cwd, "session_id": args.session_id, "end_at": args.end_at}) try: count = run( args.db, diff --git a/tests/test_capture_adapters.py b/tests/test_capture_adapters.py index e8d06510..8f9fbeaf 100644 --- a/tests/test_capture_adapters.py +++ b/tests/test_capture_adapters.py @@ -29,6 +29,191 @@ def test_inline_text_is_redacted_and_hashed() -> None: assert len(envelope.content_hash) == 64 +@pytest.mark.parametrize("extension", [None, ".md", ".html"]) +def test_mixed_generated_spans_removed_before_html_parse(capture_root, extension): + from memorymaster.core.generated_artifacts import GENERATED_CLOSER, GENERATED_OPENER + raw = f"Authored prefix.\n{GENERATED_OPENER}\nExcluded claim.\n{GENERATED_CLOSER}\nAuthored suffix." + if extension: + path = capture_root / ("mixed" + extension) + path.write_bytes(raw.encode("utf-8")) + envelope = LocalFileAdapter(path).capture() + else: + envelope = InlineTextAdapter(raw).capture() + assert "Excluded claim" not in envelope.text + assert [span.text.strip() for span in envelope.retained_spans] == ["Authored prefix.", "Authored suffix."] + assert all(raw[span.start:span.end].strip() == span.text.strip() for span in envelope.retained_spans) + + +def test_generated_only_inline_does_not_create_capture(): + from memorymaster.core.generated_artifacts import GENERATED_OPENER + with pytest.raises(CaptureRejected) as exc: + InlineTextAdapter(GENERATED_OPENER + "excluded").capture() + assert exc.value.code == "generated_artifact" + + +@pytest.mark.parametrize("tag", ["script", "style", "noscript", "template"]) +def test_hidden_html_context_survives_generated_span_removal(capture_root, tag): + from memorymaster.core.generated_artifacts import GENERATED_CLOSER, GENERATED_OPENER + + raw = (f"<{tag}>hidden before " + GENERATED_OPENER + "generated" + GENERATED_CLOSER + + f"This script-only decision must never become evidence.

Authored note.

") + path = capture_root / "hidden.html" + path.write_text(raw, encoding="utf-8") + envelope = LocalFileAdapter(path).capture() + assert envelope.text == "Authored note." + assert [span.text for span in envelope.retained_spans] == ["Authored note."] + span = envelope.retained_spans[0] + assert raw[span.start:span.end].endswith("

Authored note.

") + + +@pytest.mark.parametrize("tag", ["script", "style"]) +@pytest.mark.parametrize("fragment", ["&am", "After"), + ("&am", "p;After"), + ("", "0;After"), +]) +def test_html_token_crossing_generated_boundary_is_rejected(prefix, suffix): + from memorymaster.capture.adapters import _retained_text + from memorymaster.core.generated_artifacts import GENERATED_CLOSER, GENERATED_OPENER + + raw = "

Before

" + prefix + GENERATED_OPENER + "generated" + GENERATED_CLOSER + suffix + with pytest.raises(CaptureRejected) as exc: + _retained_text(raw, html=True) + assert exc.value.code == "ambiguous_generated_provenance" + + +def test_html_raw_text_closing_token_crossing_multiple_exclusions_is_rejected(): + from memorymaster.capture.adapters import _retained_text + from memorymaster.core.generated_artifacts import GENERATED_CLOSER, GENERATED_OPENER + + gap = GENERATED_OPENER + "generated" + GENERATED_CLOSER + raw = "