Skip to content

API reference

Package exports

akms_failure_memory

Optional deterministic failure-memory workflows for AKMS projects.

ProjectConfig dataclass 

ProjectConfig(
    source_path: Path,
    repository_id: str,
    node_namespace: str,
    domain: str,
    subdomain: str,
    paths: Mapping[str, str],
    generated: Mapping[str, str],
    validation: Mapping[str, Any],
    taxonomy: Mapping[str, Any],
    compatibility: Mapping[str, Any],
    toolchain: Mapping[str, Any],
    promotion: Mapping[str, Any],
    canonical_data: Mapping[str, Any],
    fingerprint: str,
)

Validated immutable project policy and its canonical fingerprint.

load_project_config 

load_project_config(path: str | Path) -> ProjectConfig

Load a strict TOML configuration; duplicate keys fail in tomllib.

Source code in packages/akms_failure_memory/src/akms_failure_memory/config.py 
358
359
360
361
362
363
364
365
366
367
def load_project_config(path: str | Path) -> ProjectConfig:
    """Load a strict TOML configuration; duplicate keys fail in ``tomllib``."""
    source = Path(path)
    try:
        raw = tomllib.loads(source.read_text(encoding="utf-8"))
    except (OSError, UnicodeError, tomllib.TOMLDecodeError) as exc:
        raise FailureMemoryError(
            f"Cannot load project configuration {source}: {exc}", code="config_load"
        ) from exc
    return _validate_config(raw, source.resolve())

Configuration

akms_failure_memory.config

Strict loader for failure-memory-project/v1 TOML configuration.

ProjectConfig dataclass 

ProjectConfig(
    source_path: Path,
    repository_id: str,
    node_namespace: str,
    domain: str,
    subdomain: str,
    paths: Mapping[str, str],
    generated: Mapping[str, str],
    validation: Mapping[str, Any],
    taxonomy: Mapping[str, Any],
    compatibility: Mapping[str, Any],
    toolchain: Mapping[str, Any],
    promotion: Mapping[str, Any],
    canonical_data: Mapping[str, Any],
    fingerprint: str,
)

Validated immutable project policy and its canonical fingerprint.

load_project_config 

load_project_config(path: str | Path) -> ProjectConfig

Load a strict TOML configuration; duplicate keys fail in tomllib.

Source code in packages/akms_failure_memory/src/akms_failure_memory/config.py 
358
359
360
361
362
363
364
365
366
367
def load_project_config(path: str | Path) -> ProjectConfig:
    """Load a strict TOML configuration; duplicate keys fail in ``tomllib``."""
    source = Path(path)
    try:
        raw = tomllib.loads(source.read_text(encoding="utf-8"))
    except (OSError, UnicodeError, tomllib.TOMLDecodeError) as exc:
        raise FailureMemoryError(
            f"Cannot load project configuration {source}: {exc}", code="config_load"
        ) from exc
    return _validate_config(raw, source.resolve())

Recording

akms_failure_memory.record

One canonical recorder for interactive, machine, and direct-edit workflows.

add_lesson 

add_lesson(
    *,
    config_path: str | Path,
    repository_root: str | Path,
    global_vault: str | Path,
    request_path: str | Path | None = None,
    interactive: bool = False,
    input_fn: Callable[[str], str] = input,
) -> dict[str, Any]

Allocate, validate, atomically publish, and compile one canonical record.

Source code in packages/akms_failure_memory/src/akms_failure_memory/record.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
def add_lesson(
    *,
    config_path: str | Path,
    repository_root: str | Path,
    global_vault: str | Path,
    request_path: str | Path | None = None,
    interactive: bool = False,
    input_fn: Callable[[str], str] = input,
) -> dict[str, Any]:
    """Allocate, validate, atomically publish, and compile one canonical record."""
    if interactive == (request_path is not None):
        raise FailureMemoryError(
            "Choose exactly one of interactive or request_path", code="record_request"
        )
    config = load_project_config(config_path)
    root = Path(repository_root).resolve(strict=True)
    registry_path = config.resolve(root, "registry")
    lock_path = config.resolve(root, "lock")
    if interactive:
        request = _interactive_request(input_fn)
    else:
        # Non-None here: the exactly-one check above rejects the other case.
        assert request_path is not None
        request = _load_request(Path(request_path))
    _validate_request_shape(request)
    with ProjectLock(
        lock_path, timeout_seconds=float(config.toolchain["timeout_seconds"])
    ):
        try:
            prior = registry_path.read_bytes()
            registry = json.loads(prior.decode("utf-8"))
        except (OSError, UnicodeError, json.JSONDecodeError) as exc:
            raise FailureMemoryError(
                f"Cannot load canonical registry: {exc}", code="registry_load"
            ) from exc
        lesson = {"id": _next_id(config, registry["lessons"]), **request}
        candidate = dict(registry)
        candidate["lessons"] = [*registry["lessons"], lesson]
        candidate_bytes = _canonical_registry(candidate)
        validate_registry_bytes(candidate_bytes, registry_path, config)
        _atomic_replace(registry_path, candidate_bytes)
        try:
            compilation = run_compiler(
                config_path=config_path,
                repository_root=root,
                global_vault=global_vault,
                mode="write",
            )
        except Exception:
            _atomic_replace(registry_path, prior)
            raise
    return {
        "status": "created",
        "schema_version": "failure-memory-registry/v1",
        "record": lesson,
        "source_sha256": compilation["source_sha256"],
        "affected_outputs": compilation["written"],
        "promotion": "not-performed",
    }

Compiler

akms_failure_memory.compiler

Configuration-driven deterministic lesson compiler and exact-route generator.

validate_registry_bytes 

validate_registry_bytes(
    data: bytes, source: Path, config: ProjectConfig
) -> tuple[dict[str, Any], ...]

Validate the canonical append-only v1 registry without mutating it.

Source code in packages/akms_failure_memory/src/akms_failure_memory/compiler.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
def validate_registry_bytes(
    data: bytes, source: Path, config: ProjectConfig
) -> tuple[dict[str, Any], ...]:
    """Validate the canonical append-only v1 registry without mutating it."""
    registry = _decode_registry(data, source, config)
    lessons: list[dict[str, Any]] = []
    seen: set[str] = set()
    id_pattern = re.compile(str(config.validation["id_pattern"]))
    for index, item in enumerate(registry["lessons"]):
        lesson = _closed(
            item, _LESSON_KEYS, f"lessons[{index}]", optional=frozenset({"notes"})
        )
        lesson_id = _scalar(lesson, "id", f"lessons[{index}]")
        if id_pattern.fullmatch(lesson_id) is None:
            raise FailureMemoryError(
                f"Malformed lesson ID {lesson_id!r}", code="registry_id"
            )
        if lesson_id in seen:
            raise FailureMemoryError(
                f"Duplicate lesson ID {lesson_id}", code="duplicate_id"
            )
        seen.add(lesson_id)
        for key in ("date_found", "date_fixed"):
            if _DATE_RE.fullmatch(_scalar(lesson, key, lesson_id)) is None:
                raise FailureMemoryError(
                    f"{lesson_id}.{key} is not YYYY-MM-DD", code="registry_schema"
                )
        for key in ("date_found_precision", "date_fixed_precision"):
            if _scalar(lesson, key, lesson_id) not in {"exact", "approximate"}:
                raise FailureMemoryError(
                    f"{lesson_id}.{key} is invalid", code="registry_schema"
                )
        location = _closed(lesson["location"], _LOCATION_KEYS, f"{lesson_id}.location")
        for key in _LOCATION_KEYS:
            _scalar(location, key, f"{lesson_id}.location")
        if not location["file"]:
            raise FailureMemoryError(
                f"{lesson_id}.location.file is empty", code="registry_path"
            )
        if ";" in location["file"]:
            raise FailureMemoryError(
                f"{lesson_id}: location.file must not use ';' as a multi-path delimiter",
                code="registry_path",
            )
        found = _closed(lesson["found_by"], _FOUND_KEYS, f"{lesson_id}.found_by")
        references = _closed(
            lesson["references"], _REFERENCE_KEYS, f"{lesson_id}.references"
        )
        for key in _FOUND_KEYS:
            _scalar(found, key, f"{lesson_id}.found_by")
        for key in _REFERENCE_KEYS:
            _scalar(references, key, f"{lesson_id}.references")
        for key in ("symptom", "root_cause", "fix", "prevention"):
            _scalar(lesson, key, lesson_id)
        if "notes" in lesson:
            _scalar(lesson, "notes", lesson_id)
        related = lesson["related"]
        if not isinstance(related, list) or any(
            not isinstance(value, str) for value in related
        ):
            raise FailureMemoryError(
                f"{lesson_id}.related must be a string array", code="registry_schema"
            )
        if len(related) != len(set(related)):
            raise FailureMemoryError(
                f"{lesson_id}.related contains duplicates", code="registry_schema"
            )
        lessons.append(lesson)
    for lesson in lessons:
        for related in lesson["related"]:
            if related == lesson["id"]:
                raise FailureMemoryError(
                    f"{related} cannot load itself", code="related_id"
                )
            if related not in seen:
                raise FailureMemoryError(
                    f"{lesson['id']} references unknown related lesson ID {related}",
                    code="related_id",
                )
    return tuple(lessons)

run_compiler 

run_compiler(
    *,
    config_path: str | Path,
    repository_root: str | Path,
    global_vault: str | Path | None = None,
    output_root: str | Path | None = None,
    mode: str = "write",
) -> dict[str, Any]

Validate, compare, and atomically publish configured deterministic outputs.

Source code in packages/akms_failure_memory/src/akms_failure_memory/compiler.py 
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
def run_compiler(
    *,
    config_path: str | Path,
    repository_root: str | Path,
    global_vault: str | Path | None = None,
    output_root: str | Path | None = None,
    mode: str = "write",
) -> dict[str, Any]:
    """Validate, compare, and atomically publish configured deterministic outputs."""
    if mode not in {"write", "check", "validate", "dry-run"}:
        raise FailureMemoryError(f"Unsupported compiler mode {mode!r}", code="usage")
    config = load_project_config(config_path)
    repo = Path(repository_root).resolve(strict=True)
    raw_destination = Path(output_root).expanduser() if output_root else repo
    if raw_destination.is_symlink():
        raise FailureMemoryError(
            "Output root must not be a symlink", code="path_escape"
        )
    destination_root = raw_destination.resolve(strict=False)
    # Only mode="write" may create filesystem structure. validate/check/
    # dry-run are read-only compiler modes and must create nothing --
    # including destination_root itself and the lock's parent directory
    # (create_parent_directories below) -- even when the caller passes an
    # explicit output_root that does not yet exist. The comparison logic
    # below already handles a nonexistent destination_root/generated
    # directory correctly (Path.exists() is False, everything reports as
    # "added"/missing), so no destination_root.is_dir() guard is needed.
    if mode == "write":
        destination_root.mkdir(parents=True, exist_ok=True)
    with ProjectLock(
        config.resolve(repo, "lock"),
        timeout_seconds=float(config.toolchain["timeout_seconds"]),
        create_parent_directories=(mode == "write"),
    ):
        compilation = compile_registry(config, repo)
        if mode == "validate":
            return {
                "status": "valid",
                "mode": mode,
                "source_sha256": compilation.source_sha256,
                "counts": {
                    "lessons": len(compilation.nodes),
                    "path_routes": len(compilation.canonical_routes["by_path"]),
                    "warnings": len(compilation.warnings),
                },
                "warnings": list(compilation.warnings),
                "config_fingerprint": config.fingerprint,
            }
        generated = _target(destination_root, config.paths["generated_nodes"])
        local_nodes = _target(destination_root, config.paths["local_nodes"])
        routes = _target(destination_root, config.paths["routes"])
        if global_vault is None:
            raise FailureMemoryError(
                "An explicit read-only global vault is required",
                code="global_vault_required",
            )
        _validate_collisions(
            compilation, generated, local_nodes, Path(global_vault).resolve()
        )
        expected = {
            generated / f"{node.node_id}.md": serialize_node(node)
            for node in compilation.nodes
        }
        expected[routes] = serialize_routes(compilation.adapted_routes)
        added, changed = [], []
        for path, content in expected.items():
            if not path.exists():
                added.append(path)
            elif path.read_bytes() != content:
                changed.append(path)
        expected_nodes = {path for path in expected if path.parent == generated}
        stale = (
            [
                path
                for path in generated.glob(f"{config.node_namespace}-*.md")
                if path not in expected_nodes
                and _frontmatter_id(path) == path.stem
                and config.compatibility["generated_warning"]
                in path.read_text(encoding="utf-8")
            ]
            if generated.is_dir()
            else []
        )

        def key(path: Path) -> bytes:
            return path.as_posix().encode("utf-8")

        added, changed, stale = map(
            lambda values: sorted(values, key=key), (added, changed, stale)
        )
        if mode == "write":
            _publish_transaction(destination_root, expected, added + changed, stale)
        drift = bool(added or changed or stale)

        def display(values: list[Path]) -> list[str]:
            return [path.relative_to(destination_root).as_posix() for path in values]

        status = (
            "drift"
            if mode == "check" and drift
            else "clean"
            if mode == "check"
            else "dry-run"
            if mode == "dry-run"
            else "written"
        )
        return {
            "status": status,
            "mode": mode,
            "source_sha256": compilation.source_sha256,
            "counts": {
                "lessons": len(compilation.nodes),
                "path_routes": len(compilation.canonical_routes["by_path"]),
                "added": len(added),
                "changed": len(changed),
                "stale": len(stale),
                "warnings": len(compilation.warnings),
            },
            "added": display(added),
            "changed": display(changed),
            "stale": display(stale),
            "written": display(added + changed) if mode == "write" else [],
            "warnings": list(compilation.warnings),
            "config_fingerprint": config.fingerprint,
        }

Refresh

akms_failure_memory.refresh

Pinned toolchain preflight and single-writer deterministic refresh.

preflight 

preflight(
    *, config: ProjectConfig, repository_root: str | Path
) -> dict[str, Any]

Check that the installed AKMS is usable and repo2md is callable.

This is deliberately NOT an identity check. It blocks on exactly two things:

  • the installed AKMS schema version is incompatible with the project, and
  • repo2md cannot be invoked with the export contract this package needs.

Both are conditions under which the next step genuinely cannot run.

Everything else — the AKMS version string, the AKMS public-API digest, and the repo2md checkout version / commit / cleanliness / fixture digest — is reported as an advisory and never raises. Those answer "is this the exact artifact we certified?", which matters when publishing a release, not when a developer is using the tool. Enforcing them on every resolve made ordinary, correct edits to AKMS fail closed here, so pins got chased instead of code getting fixed: the duplicated run_qmd.sh lookup in akms.graph.generate_loadout survived precisely because removing it would have tripped the public-API digest.

Advisories are returned under advisories so drift stays visible. Publication paths that do need exact identity should compare these values themselves rather than relying on this function to refuse.

Source code in packages/akms_failure_memory/src/akms_failure_memory/refresh.py 
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
def preflight(*, config: ProjectConfig, repository_root: str | Path) -> dict[str, Any]:
    """Check that the installed AKMS is usable and repo2md is callable.

    This is deliberately NOT an identity check. It blocks on exactly two things:

      * the installed AKMS schema version is incompatible with the project, and
      * repo2md cannot be invoked with the export contract this package needs.

    Both are conditions under which the next step genuinely cannot run.

    Everything else — the AKMS version string, the AKMS public-API digest, and
    the repo2md checkout version / commit / cleanliness / fixture digest — is
    reported as an advisory and never raises. Those answer "is this the exact
    artifact we certified?", which matters when publishing a release, not when a
    developer is using the tool. Enforcing them on every resolve made ordinary,
    correct edits to AKMS fail closed here, so pins got chased instead of code
    getting fixed: the duplicated run_qmd.sh lookup in
    ``akms.graph.generate_loadout`` survived precisely because removing it would
    have tripped the public-API digest.

    Advisories are returned under ``advisories`` so drift stays visible.
    Publication paths that do need exact identity should compare these values
    themselves rather than relying on this function to refuse.
    """
    root = Path(repository_root).resolve(strict=True)
    advisories: list[dict[str, Any]] = []

    def note(code: str, message: str, **details: Any) -> None:
        entry: dict[str, Any] = {"code": code, "message": message}
        if details:
            entry["details"] = details
        advisories.append(entry)

    # ── Blocking: schema compatibility ────────────────────────────────────
    # A v2 project cannot be served by an AKMS that speaks a different schema.
    if akms.AKMS_SCHEMA_VERSION != config.toolchain["akms_schema_version"]:
        raise FailureMemoryError(
            "Installed AKMS schema is incompatible", code="akms_contract"
        )

    # ── Advisory: AKMS identity ───────────────────────────────────────────
    pinned_version = config.toolchain.get("akms_version")
    if pinned_version and akms.__version__ != pinned_version:
        note(
            "akms_version_drift",
            "Installed AKMS version differs from the project pin",
            actual=akms.__version__,
            expected=pinned_version,
        )

    try:
        digest = _akms_public_digest()
    except FailureMemoryError as exc:
        digest = ""
        note("akms_public_api_unreadable", str(exc))
    pinned_digest = config.toolchain.get("akms_public_api_sha256")
    if pinned_digest and digest and digest != pinned_digest:
        note(
            "akms_public_api_drift",
            "Installed AKMS public API differs from the project pin",
            actual=digest,
            expected=pinned_digest,
        )

    # ── Blocking: repo2md must actually be callable ───────────────────────
    command = _resolve_command(tuple(config.toolchain["repo2md_command"]))
    contract = _run(
        [*command, "export-akms", "--help"],
        cwd=root,
        timeout=float(config.toolchain["timeout_seconds"]),
    )
    contract_text = contract.stdout + contract.stderr
    if contract.returncode != 0 or any(
        flag not in contract_text for flag in ("--output", "--phase", "--json")
    ):
        raise FailureMemoryError(
            "repo2md export-akms CLI contract is unavailable", code="repo2md_contract"
        )

    # ── Advisory: repo2md identity, best effort ───────────────────────────
    distribution = _installed_distribution(command[0])
    metadata_version, editable_root = distribution if distribution else ("", None)
    if distribution is None:
        note(
            "repo2md_distribution_unknown",
            "repo2md is not linked to a verifiable editable distribution",
        )

    configured_root = config.toolchain.get("repo2md_root")
    environment_root = os.environ.get("AKMS_REPO2MD_ROOT")
    tool_root: Path | None
    try:
        if configured_root:
            tool_root = root.joinpath(
                *PurePosixPath(str(configured_root)).parts
            ).resolve(strict=True)
        elif environment_root:
            tool_root = Path(environment_root).expanduser().resolve(strict=True)
        else:
            tool_root = editable_root
    except OSError as exc:
        tool_root = editable_root
        note(
            "repo2md_root_unresolved", f"Cannot resolve configured repo2md root: {exc}"
        )

    if (
        tool_root is not None
        and editable_root is not None
        and tool_root != editable_root
    ):
        note(
            "repo2md_checkout_mismatch",
            "repo2md executable is not linked to the configured checkout",
            configured=str(tool_root),
            linked=str(editable_root),
        )

    checkout_version = ""
    observed_commit = ""
    observed_dirty = False
    export_schema = config.toolchain.get("repo2md_export_schema_version")
    fixture_sha = ""

    if tool_root is not None:
        try:
            checkout_version = _checkout_version(tool_root)
        except Exception as exc:  # noqa: BLE001 - advisory only
            note("repo2md_version_unknown", f"Cannot read repo2md version: {exc}")
        expected_version = str(config.toolchain["repo2md_version"])
        if checkout_version and checkout_version != expected_version:
            note(
                "repo2md_version_drift",
                "repo2md version differs from the project pin",
                actual=checkout_version,
                expected=expected_version,
            )

        head = _run(
            ["git", "-C", str(tool_root), "rev-parse", "HEAD"],
            cwd=root,
            timeout=float(config.toolchain["timeout_seconds"]),
        )
        observed_commit = head.stdout.strip()
        if head.returncode != 0:
            note("repo2md_commit_unknown", "Cannot read repo2md checkout commit")
        elif observed_commit != config.toolchain["repo2md_commit"]:
            note(
                "repo2md_commit_drift",
                "repo2md checkout is not at the pinned commit",
                actual=observed_commit,
                expected=config.toolchain["repo2md_commit"],
            )

        dirty = _run(
            ["git", "-C", str(tool_root), "status", "--porcelain"],
            cwd=root,
            timeout=float(config.toolchain["timeout_seconds"]),
        )
        if dirty.returncode != 0:
            note("repo2md_dirty_unknown", "Cannot inspect repo2md checkout state")
        else:
            observed_dirty = bool(dirty.stdout)
            if (
                config.toolchain["repo2md_dirty_policy"] == "require-clean"
                and observed_dirty
            ):
                note("repo2md_dirty", "repo2md checkout has uncommitted changes")

        try:
            pin_version, observed_schema, fixture_sha = _fixture_identity(tool_root)
            export_schema = observed_schema
            if checkout_version and pin_version != checkout_version:
                note(
                    "repo2md_pin_version_drift",
                    "repo2md integration pin version does not match the checkout",
                    actual=pin_version,
                    expected=checkout_version,
                )
            if observed_schema != config.toolchain["repo2md_export_schema_version"]:
                note(
                    "repo2md_export_schema_drift",
                    "repo2md export schema differs from the project pin",
                    actual=observed_schema,
                    expected=config.toolchain["repo2md_export_schema_version"],
                )
            pinned_fixture = config.toolchain.get("repo2md_fixture_sha256")
            if pinned_fixture and fixture_sha != pinned_fixture:
                note(
                    "repo2md_fixture_drift",
                    "repo2md fixture digest differs from the project pin",
                    actual=fixture_sha,
                    expected=pinned_fixture,
                )
        except FailureMemoryError as exc:
            note("repo2md_fixture_unreadable", str(exc))

    if metadata_version and checkout_version and metadata_version != checkout_version:
        # Editable installers may retain stale wheel metadata; the checkout is
        # the observed authority.
        metadata_status = "stale"
    else:
        metadata_status = "current"

    return {
        "status": "ok",
        "advisories": advisories,
        "akms": {
            "version": akms.__version__,
            "schema": akms.AKMS_SCHEMA_VERSION,
            "public_api_sha256": digest,
        },
        "repo2md": {
            "command": list(command),
            "checkout": str(tool_root) if tool_root is not None else "",
            "version": checkout_version,
            "distribution_version": metadata_version,
            "distribution_metadata": metadata_status,
            "export_schema_version": export_schema,
            "fixture_sha256": fixture_sha,
            "commit": observed_commit,
            "dirty": observed_dirty,
        },
    }

refresh_project 

refresh_project(
    *,
    action: str,
    config_path: str | Path,
    repository_root: str | Path,
    global_vault: str | Path,
    phase: int = 1,
    generated_at: str | None = None,
    force_lock: bool = False,
) -> dict[str, Any]

Run one refresh stage or the ordered lessons→mirror→graph chain.

Source code in packages/akms_failure_memory/src/akms_failure_memory/refresh.py 
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
def refresh_project(
    *,
    action: str,
    config_path: str | Path,
    repository_root: str | Path,
    global_vault: str | Path,
    phase: int = 1,
    generated_at: str | None = None,
    force_lock: bool = False,
) -> dict[str, Any]:
    """Run one refresh stage or the ordered lessons→mirror→graph chain."""
    config = load_project_config(config_path)
    root = Path(repository_root).resolve(strict=True)
    vault = Path(global_vault).resolve(strict=True)
    if action == "preflight":
        return preflight(config=config, repository_root=root)
    if action == "status":
        return status(config, root)
    with ProjectLock(
        config.resolve(root, "lock"),
        timeout_seconds=float(config.toolchain["timeout_seconds"]),
        force_stale=force_lock,
    ):
        if action == "clean":
            return _clean(config, root)
        identity = preflight(config=config, repository_root=root)
        stages = {}
        if action in {"lessons", "all"}:
            stages["lessons"] = run_compiler(
                config_path=config_path,
                repository_root=root,
                global_vault=vault,
                mode="write",
            )
        if action in {"mirror", "all"}:
            stages["mirror"] = _mirror(
                config, root, phase=phase, generated_at=generated_at, identity=identity
            )
        if action in {"graph", "all"}:
            stages["graph"] = _graph(config, root, vault, generated_at=generated_at)
        if not stages:
            raise FailureMemoryError(f"Unknown refresh action {action!r}", code="usage")
        return {
            "status": "ok",
            "action": action,
            "stages": stages,
            "toolchain": identity,
            "config_fingerprint": config.fingerprint,
        }

Provider

akms_failure_memory.provider

Harness-neutral deterministic failure-memory provider contract.

load_provider_request 

load_provider_request(
    source: str | Path | Mapping[str, Any],
) -> dict[str, Any]

Load and strictly validate one closed v1 provider request.

Source code in packages/akms_failure_memory/src/akms_failure_memory/provider.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
def load_provider_request(source: str | Path | Mapping[str, Any]) -> dict[str, Any]:
    """Load and strictly validate one closed v1 provider request."""
    if isinstance(source, Mapping):
        raw = dict(source)
    else:
        try:
            raw = json.loads(Path(source).read_text(encoding="utf-8"))
        except (OSError, json.JSONDecodeError) as exc:
            raise FailureMemoryError(
                f"Cannot load provider request: {exc}", code="provider_request"
            ) from exc
    if not isinstance(raw, dict):
        raise FailureMemoryError(
            "Provider request root must be an object", code="provider_request"
        )
    missing = _REQUIRED_REQUEST_FIELDS - raw.keys()
    extra = raw.keys() - _REQUEST_FIELDS
    if missing or extra:
        raise FailureMemoryError(
            f"Invalid provider request fields; missing={sorted(missing)}, unexpected={sorted(extra)}",
            code="provider_request",
        )
    if raw["schema_version"] != REQUEST_SCHEMA_VERSION:
        raise FailureMemoryError(
            "Unsupported provider request schema", code="schema_version"
        )
    mode = _text(raw["mode"], "mode")
    if mode not in {"pre-task", "post-diff"}:
        raise FailureMemoryError(
            "mode must be pre-task or post-diff", code="provider_request"
        )
    role = _text(raw["role"], "role")
    if mode == "pre-task" and role not in _IMPLEMENTER_ROLES:
        raise FailureMemoryError(
            "pre-task mode requires implementer role", code="provider_request"
        )
    if mode == "post-diff" and role not in _REVIEWER_ROLES:
        raise FailureMemoryError(
            "post-diff mode requires a reviewer role", code="provider_request"
        )
    refresh_policy = _text(raw["refresh_policy"], "refresh_policy")
    if refresh_policy not in {"never", "require-current"}:
        raise FailureMemoryError(
            "refresh_policy must be never or require-current", code="provider_request"
        )
    changed = _paths(raw.get("changed_paths", []), "changed_paths")
    base = raw.get("base")
    head = raw.get("head")
    if base is not None:
        base = _text(base, "base")
    if head is not None:
        head = _text(head, "head")
    if changed and base is not None:
        raise FailureMemoryError(
            "Provide changed_paths or base/head, not both", code="provider_request"
        )
    task = raw["task"]
    if not isinstance(task, dict):
        raise FailureMemoryError("task must be an object", code="provider_request")
    normalized = {
        "schema_version": REQUEST_SCHEMA_VERSION,
        "invocation_id": _identifier(raw["invocation_id"], "invocation_id"),
        "repository_id": _identifier(raw["repository_id"], "repository_id"),
        "baseline": _text(raw["baseline"], "baseline"),
        "mode": mode,
        "role": role,
        "declared_paths": list(_paths(raw["declared_paths"], "declared_paths")),
        "changed_paths": list(changed),
        "base": base,
        "head": head,
        "refresh_policy": refresh_policy,
        "output_dir": _path(raw["output_dir"], "output_dir"),
        "task": json.loads(json.dumps(task, ensure_ascii=False)),
    }
    if not normalized["declared_paths"]:
        raise FailureMemoryError(
            "declared_paths must not be empty", code="provider_request"
        )
    return normalized

validate_publication 

validate_publication(
    *, config_path: str | Path, repository_root: str | Path
) -> list[str]

Package-owned deterministic validation of the published graph.

Two published artifacts must agree: the graph and the generated project node files it was built from. This verifies what is there — it never re-derives the compiler's or the graph builder's rules — and it is the single definition both consumers inherit through validate_fingerprint, so neither has to (and neither may) mirror a package predicate.

Checks, all against this project's own namespace only:

  1. The graph carries this project's identity: graph.repo_id equals the configured repository_id. Fingerprint reproducibility alone cannot see this — a graph published under a foreign identity reproduces consistently.
  2. The project-namespaced nodes in the graph are exactly the published generated node files — a graph node with no published file behind it is fabricated content, and a published file missing from the graph is the ordinary "nodes were recompiled but the graph was not rebuilt" state.
  3. Every field the two artifacts share agrees, and every load_with target a published node names is present in the graph, because resolution will read that content and try to load those targets.

Returns a list of human-readable problems; an empty list means the publication is internally consistent and carries this project's identity.

Source code in packages/akms_failure_memory/src/akms_failure_memory/provider.py 
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
def validate_publication(
    *, config_path: str | Path, repository_root: str | Path
) -> list[str]:
    """Package-owned deterministic validation of the published graph.

    Two published artifacts must agree: the graph and the generated project
    node files it was built from. This verifies what is there — it never
    re-derives the compiler's or the graph builder's rules — and it is the
    single definition both consumers inherit through ``validate_fingerprint``,
    so neither has to (and neither may) mirror a package predicate.

    Checks, all against this project's own namespace only:

    1. The graph carries this project's identity: ``graph.repo_id`` equals the
       configured ``repository_id``. Fingerprint reproducibility alone cannot
       see this — a graph published under a foreign identity reproduces
       consistently.
    2. The project-namespaced nodes in the graph are exactly the published
       generated node files — a graph node with no published file behind it is
       fabricated content, and a published file missing from the graph is the
       ordinary "nodes were recompiled but the graph was not rebuilt" state.
    3. Every field the two artifacts share agrees, and every ``load_with``
       target a published node names is present in the graph, because
       resolution will read that content and try to load those targets.

    Returns a list of human-readable problems; an empty list means the
    publication is internally consistent and carries this project's identity.
    """
    config = load_project_config(config_path)
    root = Path(repository_root).resolve(strict=True)
    graph_path = config.resolve(root, "graph")
    try:
        document = json.loads(graph_path.read_text(encoding="utf-8"))
        metadata = document["graph"]
        raw_nodes = document["nodes"]
        if not isinstance(metadata, dict) or not isinstance(raw_nodes, list):
            raise TypeError("graph document shape is invalid")
    except (OSError, UnicodeError, KeyError, TypeError, json.JSONDecodeError):
        return [f"published graph is unreadable: {config.paths['graph']}"]
    problems: list[str] = []
    if metadata.get("repo_id") != config.repository_id:
        problems.append(
            f"published graph repo_id {metadata.get('repo_id')!r} does not match "
            f"the configured repository_id {config.repository_id!r}"
        )
    graph_nodes = {
        node["id"]: node
        for node in raw_nodes
        if isinstance(node, dict) and isinstance(node.get("id"), str)
    }
    prefix = f"{config.node_namespace}-"
    generated = config.resolve(root, "generated_nodes")
    published_ids = (
        sorted(path.stem for path in generated.glob(f"{prefix}*.md"))
        if generated.is_dir()
        else []
    )
    graph_project_ids = sorted(
        node_id for node_id in graph_nodes if node_id.startswith(prefix)
    )
    for node_id in sorted(set(graph_project_ids) - set(published_ids)):
        problems.append(
            f"graph node {node_id} has no published generated node file behind it"
        )
    for node_id in sorted(set(published_ids) - set(graph_project_ids)):
        problems.append(f"published generated node {node_id} is missing from the graph")
    for node_id in sorted(set(published_ids) & set(graph_project_ids)):
        front = _published_frontmatter(generated / f"{node_id}.md")
        if front is None:
            problems.append(
                f"published generated node {node_id} has unreadable front matter"
            )
            continue
        graph_node = graph_nodes[node_id]
        disagreeing = sorted(
            key
            for key in set(front) & set(graph_node)
            if not _fields_agree(front[key], graph_node[key])
        )
        if disagreeing:
            problems.append(
                f"graph node {node_id} disagrees with its published generated "
                f"file on: {', '.join(disagreeing)}"
            )
        for target in front.get("load_with") or []:
            if target not in graph_nodes:
                problems.append(
                    f"load_with target {target} named by published node "
                    f"{node_id} is missing from the graph"
                )
    return problems

resolve_provider 

resolve_provider(
    *,
    config_path: str | Path,
    repository_root: str | Path,
    request_source: str | Path | Mapping[str, Any],
    write_artifacts: bool = True,
) -> dict[str, Any]

Resolve one provider request using only pinned public AKMS services.

Source code in packages/akms_failure_memory/src/akms_failure_memory/provider.py 
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
def resolve_provider(
    *,
    config_path: str | Path,
    repository_root: str | Path,
    request_source: str | Path | Mapping[str, Any],
    write_artifacts: bool = True,
) -> dict[str, Any]:
    """Resolve one provider request using only pinned public AKMS services."""
    config = load_project_config(config_path)
    root = Path(repository_root).resolve(strict=True)
    request = load_provider_request(request_source)
    if request["repository_id"] != config.repository_id:
        raise FailureMemoryError(
            "repository_id does not match project config", code="provider_identity"
        )
    # write_artifacts is the caller's declaration of intent to mutate the
    # target repository (provider evidence outputs). A read-only caller
    # (write_artifacts=False -- e.g. a consumer's strictly read-only
    # surface) must not have lock acquisition itself create filesystem
    # structure; see ProjectLock.acquire and locks.py for the fail-closed
    # behavior when the lock's parent directory does not yet exist.
    with ProjectLock(
        config.resolve(root, "lock"),
        timeout_seconds=float(config.toolchain["timeout_seconds"]),
        create_parent_directories=write_artifacts,
    ):
        if request["refresh_policy"] == "require-current":
            _require_current(config, root)
        return _resolve_provider_locked(
            config=config,
            root=root,
            request=request,
            write_artifacts=write_artifacts,
        )

validate_fingerprint 

validate_fingerprint(
    *,
    config_path: str | Path,
    repository_root: str | Path,
    request_source: str | Path | Mapping[str, Any],
    result_path: str | Path,
) -> dict[str, Any]

Recompute a result fingerprint without writing provider artifacts.

current requires BOTH that the recomputed fingerprint matches the recorded one AND that :func:validate_publication finds the published graph valid. Fingerprint reproducibility alone cannot notice a graph published under a foreign repository identity, or a graph whose project-namespaced nodes disagree with the published generated node files — a consistently-wrong graph reproduces consistently — so those states report stale (republish the graph), never current.

Source code in packages/akms_failure_memory/src/akms_failure_memory/provider.py 
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
def validate_fingerprint(
    *,
    config_path: str | Path,
    repository_root: str | Path,
    request_source: str | Path | Mapping[str, Any],
    result_path: str | Path,
) -> dict[str, Any]:
    """Recompute a result fingerprint without writing provider artifacts.

    ``current`` requires BOTH that the recomputed fingerprint matches the
    recorded one AND that :func:`validate_publication` finds the published
    graph valid. Fingerprint reproducibility alone cannot notice a graph
    published under a foreign repository identity, or a graph whose
    project-namespaced nodes disagree with the published generated node files
    — a consistently-wrong graph reproduces consistently — so those states
    report ``stale`` (republish the graph), never ``current``.
    """
    try:
        prior = json.loads(Path(result_path).read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        raise FailureMemoryError(
            f"Cannot load provider result: {exc}", code="provider_result"
        ) from exc
    if (
        not isinstance(prior, dict)
        or prior.get("schema_version") != RESULT_SCHEMA_VERSION
    ):
        raise FailureMemoryError(
            "Unsupported provider result schema", code="provider_result"
        )
    current = resolve_provider(
        config_path=config_path,
        repository_root=repository_root,
        request_source=request_source,
        write_artifacts=False,
    )
    stale = prior.get("fingerprint") != current["fingerprint"] or bool(
        validate_publication(config_path=config_path, repository_root=repository_root)
    )
    return {
        "status": "stale" if stale else "current",
        "stale": stale,
        "recorded_fingerprint": prior.get("fingerprint"),
        "current_fingerprint": current["fingerprint"],
    }