Coverage for astrocyte/okf.py: 98%
242 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-10-29 22:10 +0000
1"""OKF v0.2 bundle export.
3Projects compiled wiki pages into an Open Knowledge Format bundle: a directory
4of markdown files with YAML frontmatter, plus ``index.md`` / ``log.md``.
6Spec: ``GoogleCloudPlatform/knowledge-catalog/okf/SPEC.md`` (v0.2).
8This is a **projection**, not a storage format. Postgres remains the system of
9record; nothing here writes back. It sits beside :mod:`astrocyte.portability`
10(AMA), which exports raw memories from ``recall()`` results and structurally
11cannot see wiki pages.
13Fields are emitted only where real data exists. ``type`` is the sole required
14frontmatter key (SPEC 4.1, 11), so a sparse bundle is fully conformant, and
15consumers MUST tolerate absent optional families. Deliberate omissions:
17``verified``
18 Astrocyte has no review workflow, so no record written today carries
19 verification events and consumers correctly derive the **unverified** tier
20 (SPEC 5.3). :func:`verified_for` reads ``metadata["_verified"]`` so the
21 export is complete the moment such a workflow exists — but nothing is
22 invented in the meantime.
23``status``
24 No lifecycle column exists on any record. Absent ``status`` already means
25 ``stable`` (SPEC 5.4), which is accurate for a compiled page.
26``stale_after``
27 Omitted for wiki pages, which no TTL governs. **Emitted for memories** when
28 lifecycle is enabled, derived from ``delete_after_days`` — see
29 :func:`stale_after_for`.
30``description``
31 ``WikiPage`` has no summary field; ``wiki_revisions.summary`` receives the
32 title. Emitting it would duplicate ``title`` rather than describe the page.
33"""
35from __future__ import annotations
37import re
38from dataclasses import dataclass, field
39from datetime import datetime, timedelta
40from pathlib import Path
41from typing import TYPE_CHECKING, Any
43import yaml
45from .portability import _safe_resolve
47if TYPE_CHECKING: # pragma: no cover
48 from .types import VectorItem, WikiPage
50OKF_VERSION = "0.2"
52# SPEC 3.1: reserved at every level of the hierarchy.
53_RESERVED_STEMS = frozenset({"index", "log"})
55# WikiPage.kind -> OKF `type`. Types are not centrally registered (SPEC 4.1);
56# these are descriptive and self-explanatory, which is all the spec asks.
57_KIND_TO_TYPE = {"topic": "Topic", "entity": "Entity", "concept": "Concept"}
60@dataclass
61class BundleFile:
62 """One rendered file, relative to the bundle root."""
64 path: str
65 content: str
68@dataclass
69class ExportResult:
70 bank_id: str
71 concept_count: int
72 files: list[str] = field(default_factory=list)
73 skipped: list[str] = field(default_factory=list)
76def _slug_segment(raw: str) -> str:
77 """Lowercase, URL-safe path segment. Never empty, never reserved."""
78 seg = re.sub(r"[^a-z0-9._-]+", "-", str(raw).strip().lower()).strip("-.")
79 if not seg or set(seg) <= {".", "-"}:
80 seg = "concept"
81 if seg in _RESERVED_STEMS:
82 seg = f"{seg}-concept"
83 return seg
86def path_for_page_id(page_id: str, *, fallback_kind: str = "concept") -> str:
87 """Bundle-relative path for a ``page_id``, ``.md`` included.
89 ``page_id`` is already namespaced (``topic:incident-response``,
90 ``entity:alice``, ``obs:{doc}:{slug}``). Colons become directory
91 separators, which turns the existing namespace into OKF's hierarchy for
92 free — the concept ID is then the path minus ``.md`` (SPEC 2).
93 """
94 parts = [p for p in str(page_id or "").split(":") if p.strip()]
95 if not parts:
96 parts = [str(fallback_kind or "concept"), "untitled"]
97 return "/".join(_slug_segment(p) for p in parts) + ".md"
100def concept_path(page: WikiPage) -> str:
101 """Bundle-relative path for a page, ``.md`` included."""
102 return path_for_page_id(page.page_id, fallback_kind=str(page.kind or "concept"))
105def render_related(cross_links: list[str] | None) -> str:
106 """A ``## Related`` section carrying ``cross_links`` as concept links.
108 OKF expresses the concept graph as markdown links in the body, not as a
109 frontmatter field (SPEC 6.1); the bundle-relative (leading ``/``) form is
110 recommended because it survives a document moving within its directory.
111 Targets are not checked: a link to a page that has not been compiled yet
112 is well-formed, and consumers MUST tolerate it.
113 """
114 targets = [str(c).strip() for c in (cross_links or []) if str(c).strip()]
115 if not targets:
116 return ""
117 lines = ["## Related", ""]
118 lines += [f"* [{t}](/{path_for_page_id(t)})" for t in dict.fromkeys(targets)]
119 return "\n".join(lines) + "\n"
122def _iso(value: Any) -> str | None:
123 if isinstance(value, datetime):
124 return value.isoformat()
125 if isinstance(value, str) and value.strip():
126 return value.strip()
127 return None
130def _memory_resource(bank_id: str, memory_id: str) -> str:
131 """A `resource` for an internal memory row.
133 ``resource`` is REQUIRED within a sources entry (SPEC 5.1) but may name
134 something the consumer cannot follow. These IDs are meaningless outside
135 the owning bank, so the URI carries the bank to stay unambiguous.
136 """
137 return f"astrocyte://bank/{bank_id}/memory/{memory_id}"
140def build_frontmatter(page: WikiPage, *, producer: str) -> dict[str, Any]:
141 """Frontmatter for one concept. Only fields with real data are included."""
142 fm: dict[str, Any] = {"type": _KIND_TO_TYPE.get(str(page.kind), "Concept")}
144 title = (page.title or "").strip()
145 if title:
146 fm["title"] = title
148 tags = [t for t in (page.tags or []) if str(t).strip()]
149 if tags:
150 fm["tags"] = tags
152 sources = [
153 {"id": str(sid), "resource": _memory_resource(page.bank_id, str(sid))}
154 for sid in (page.source_ids or [])
155 if str(sid).strip()
156 ]
157 if sources:
158 fm["sources"] = sources
160 # `by` is REQUIRED within `generated` (SPEC 5.2), so the block is all or
161 # nothing. Astrocyte's pipeline genuinely is the producer, and
162 # `<producer>/<version>` is the actor form for tools (SPEC 7) — this is
163 # not a stand-in for the human/model actor we do not persist.
164 at = _iso(getattr(page, "revised_at", None))
165 fm["generated"] = {"by": producer, "at": at} if at else {"by": producer}
167 if page.scope and str(page.scope).strip():
168 fm["scope"] = str(page.scope).strip()
169 if page.revision:
170 fm["revision"] = int(page.revision)
172 return fm
175def _lifecycle_created_at(item: VectorItem) -> datetime | None:
176 """The creation instant lifecycle actually uses.
178 ``run_lifecycle`` reads ``metadata["_created_at"]`` rather than the typed
179 ``VectorItem.retained_at``. A published ``stale_after`` is only honest if it
180 is computed from the same value that will drive the deletion, so this
181 deliberately mirrors that read — including its fallback to nothing when the
182 key is absent, which is exactly when lifecycle cannot age the row either.
183 """
184 meta = item.metadata or {}
185 raw = meta.get("_created_at")
186 if isinstance(raw, datetime):
187 return raw
188 if isinstance(raw, str) and raw.strip():
189 try:
190 return datetime.fromisoformat(raw)
191 except ValueError:
192 return None
193 return None
196def _lifecycle_tags(item: VectorItem) -> list[str]:
197 """Tags as lifecycle sees them (``metadata["_tags"]``, comma-joined)."""
198 meta = item.metadata or {}
199 raw = meta.get("_tags")
200 if isinstance(raw, str) and raw.strip():
201 return [t.strip() for t in raw.split(",") if t.strip()]
202 return []
205def stale_after_for(item: VectorItem, lifecycle: Any, *, now: datetime | None = None) -> str | None:
206 """Absolute expiry instant for a memory, or ``None`` to omit the field.
208 OKF's ``stale_after`` is an absolute instant, not a TTL (SPEC 5.5), which
209 maps cleanly onto ``delete_after_days`` because that threshold is measured
210 from creation and therefore does not move.
212 Three rules keep the published value truthful:
214 - Only when lifecycle is **enabled**; otherwise no expiry is in force.
215 - From ``delete_after_days``, never ``archive_after_days`` — the latter is
216 measured from ``last_recalled_at`` and so shifts on every read, which is
217 not something that can be published as a fixed instant.
218 - Omitted for ``exempt_tags`` memories, which genuinely never expire.
219 """
220 del now # Absolute; independent of when the export runs.
221 if lifecycle is None or not getattr(lifecycle, "enabled", False):
222 return None
223 ttl = getattr(lifecycle, "ttl", None)
224 if ttl is None:
225 return None
227 exempt = getattr(ttl, "exempt_tags", None) or []
228 if exempt and set(_lifecycle_tags(item)) & set(exempt):
229 return None
231 created = _lifecycle_created_at(item)
232 if created is None:
233 return None
235 days = getattr(ttl, "delete_after_days", None)
236 if not isinstance(days, int) or days <= 0:
237 return None
238 return (created + timedelta(days=days)).isoformat()
241# Memory layers whose text the pipeline synthesises rather than the caller
242# supplying it. `generated.by` records who produced the *content* (SPEC 5.2),
243# so these must be attributed to the producer even when a human triggered the
244# write — crediting a consolidated observation to `human:calvin` would claim a
245# person wrote words the LLM actually wrote.
246_DERIVED_LAYERS = frozenset({"observation", "model", "compiled"})
248# ActorIdentity.type -> OKF actor prefix (SPEC 7). Consumers key trust off the
249# `human:` prefix, so a real person must map to it and a machine must not.
250_ACTOR_TYPE_TO_OKF = {"user": "human", "service": "process", "agent": "agent"}
253def okf_actor(principal: str | None) -> str | None:
254 """Map an Astrocyte principal (``{type}:{id}``) to an OKF actor (SPEC 7).
256 ``user:`` becomes ``human:`` because that prefix is what consumers key trust
257 tiers off. ``service:`` becomes ``process:``. Anything unparseable returns
258 ``None`` so the caller falls back to the producer rather than emitting a
259 malformed actor.
260 """
261 raw = (principal or "").strip()
262 if not raw or ":" not in raw:
263 return None
264 kind, _, ident = raw.partition(":")
265 ident = ident.strip()
266 if not ident:
267 return None
268 return f"{_ACTOR_TYPE_TO_OKF.get(kind.strip(), kind.strip())}:{ident}"
271def verified_for(item: VectorItem) -> list[dict[str, str]] | None:
272 """Verification events for a memory, or ``None``.
274 OKF's ``verified`` is a list of ``{by, at}`` events (SPEC 5.2), and the
275 trust tier derives from it (SPEC 5.3). **Astrocyte has no review or
276 verification workflow**, so this returns ``None`` for every record written
277 today, and consumers correctly see the *unverified* tier.
279 The reader exists so that the export is complete the moment such a workflow
280 does: write ``metadata["_verified"]`` as a list of ``{"by": <principal>,
281 "at": <ISO 8601>}`` and it flows through. Entries missing either field, or
282 carrying an unmappable actor, are dropped rather than guessed at.
283 """
284 raw = (item.metadata or {}).get("_verified")
285 if isinstance(raw, dict): # SPEC 5.2 allows a bare mapping for one verifier.
286 raw = [raw]
287 if not isinstance(raw, list):
288 return None
289 out: list[dict[str, str]] = []
290 for entry in raw:
291 if not isinstance(entry, dict):
292 continue
293 by = okf_actor(str(entry.get("by", "")))
294 at = _iso(entry.get("at"))
295 if by and at:
296 out.append({"by": by, "at": at})
297 return out or None
300def memory_concept_path(item: VectorItem) -> str:
301 """Bundle path for a memory concept.
303 Memories are flat and id-keyed, so the hierarchy comes from the kind of
304 memory rather than from the id, keeping ``memory/`` browsable.
305 """
306 group = str(item.fact_type or item.memory_layer or "memory").strip() or "memory"
307 return f"memory/{_slug_segment(group)}/{_slug_segment(item.id)}.md"
310def build_memory_frontmatter(
311 item: VectorItem,
312 *,
313 producer: str,
314 stale_after: str | None = None,
315) -> dict[str, Any]:
316 """Frontmatter for one memory concept."""
317 kind = str(item.fact_type or item.memory_layer or "memory").strip() or "memory"
318 fm: dict[str, Any] = {"type": kind.replace("_", " ").title()}
320 tags = [t for t in (item.tags or []) if str(t).strip()]
321 if tags:
322 fm["tags"] = tags
324 if item.chunk_id:
325 fm["sources"] = [
326 {
327 "id": str(item.chunk_id),
328 "resource": f"astrocyte://bank/{item.bank_id}/chunk/{item.chunk_id}",
329 }
330 ]
332 # Attribute to the writing actor only where the caller supplied the text.
333 # Derived layers are the pipeline's own words (see _DERIVED_LAYERS).
334 actor = None
335 if str(item.memory_layer or "").strip() not in _DERIVED_LAYERS:
336 actor = okf_actor((item.metadata or {}).get("_actor"))
337 by = actor or producer
339 at = _iso(item.retained_at)
340 fm["generated"] = {"by": by, "at": at} if at else {"by": by}
342 verified = verified_for(item)
343 if verified:
344 fm["verified"] = verified
346 if stale_after:
347 fm["stale_after"] = stale_after
349 # Valid time, distinct from `generated.at` (system time). OKF has no slot
350 # for the second axis, and producers MAY add keys (SPEC 4.1), so the
351 # bitemporality survives the projection instead of being flattened away.
352 occurred = _iso(item.occurred_at)
353 if occurred:
354 fm["occurred_at"] = occurred
355 if item.memory_layer:
356 fm["memory_layer"] = str(item.memory_layer)
358 return fm
361def render_memory(
362 item: VectorItem,
363 *,
364 producer: str,
365 stale_after: str | None = None,
366) -> str:
367 fm = yaml.safe_dump(
368 build_memory_frontmatter(item, producer=producer, stale_after=stale_after),
369 sort_keys=False,
370 allow_unicode=True,
371 default_flow_style=False,
372 ).strip()
373 return f"---\n{fm}\n---\n\n{(item.text or '').strip()}\n"
376def render_concept(page: WikiPage, *, producer: str) -> str:
377 fm = yaml.safe_dump(
378 build_frontmatter(page, producer=producer),
379 sort_keys=False,
380 allow_unicode=True,
381 default_flow_style=False,
382 ).strip()
383 body = (page.content or "").strip()
384 related = render_related(getattr(page, "cross_links", None))
385 if related:
386 body = f"{body}\n\n{related}" if body else related
387 return f"---\n{fm}\n---\n\n{body}\n"
390def render_index(entries: list[tuple[str, str]], *, heading: str, root: bool) -> str:
391 """An `index.md` for one directory (SPEC 8).
393 Index files carry no frontmatter, except a root index which MAY declare
394 ``okf_version``.
395 """
396 lines: list[str] = []
397 if root:
398 lines += ["---", f"okf_version: {OKF_VERSION}", "---", ""]
399 lines.append(f"# {heading}")
400 lines.append("")
401 for title, href in sorted(entries, key=lambda e: e[0].lower()):
402 lines.append(f"* [{title}]({href})")
403 lines.append("")
404 return "\n".join(lines)
407def render_log(pages: list[WikiPage]) -> str:
408 """A root `log.md` grouped by date, newest first (SPEC 9).
410 Entry verb follows the revision number: revision 1 is a Creation, any
411 later revision an Update. Grouping uses ``revised_at``, so this reflects
412 the current state of each page rather than a full revision history —
413 prior revisions live in ``astrocyte_wiki_revisions`` and are not loaded
414 here.
415 """
416 by_date: dict[str, list[str]] = {}
417 for page in pages:
418 stamp = getattr(page, "revised_at", None)
419 day = stamp.date().isoformat() if isinstance(stamp, datetime) else "unknown"
420 verb = "Creation" if int(page.revision or 1) <= 1 else "Update"
421 title = (page.title or page.page_id or "untitled").strip()
422 by_date.setdefault(day, []).append(f"* **{verb}**: [{title}]({concept_path(page)})")
424 lines = ["# Update Log", ""]
425 for day in sorted(by_date, reverse=True):
426 lines.append(f"## {day}")
427 lines.extend(sorted(by_date[day]))
428 lines.append("")
429 return "\n".join(lines)
432def build_bundle(
433 pages: list[WikiPage],
434 *,
435 bank_id: str,
436 producer: str | None = None,
437 memories: list[VectorItem] | None = None,
438 lifecycle: Any = None,
439) -> list[BundleFile]:
440 """Render a full bundle in memory. Pure — performs no I/O.
442 ``memories`` are exported as concepts under ``memory/``. When ``lifecycle``
443 is supplied and enabled, each memory carries a derived ``stale_after``
444 (see :func:`stale_after_for`).
445 """
446 if producer is None:
447 producer = _default_producer()
449 files: list[BundleFile] = []
450 seen: set[str] = set()
451 # Directory -> (title, href) for index generation.
452 dirs: dict[str, list[tuple[str, str]]] = {}
453 exported: list[WikiPage] = []
455 for page in pages:
456 path = concept_path(page)
457 if path in seen:
458 # page_id is UNIQUE per bank, so a collision means two IDs
459 # slugified together. Skip rather than silently overwrite.
460 continue
461 seen.add(path)
462 exported.append(page)
463 files.append(BundleFile(path=path, content=render_concept(page, producer=producer)))
465 parent = str(Path(path).parent)
466 parent = "" if parent == "." else parent
467 title = (page.title or page.page_id or "untitled").strip()
468 dirs.setdefault(parent, []).append((title, Path(path).name))
470 for item in memories or []:
471 path = memory_concept_path(item)
472 if path in seen:
473 continue
474 seen.add(path)
475 files.append(
476 BundleFile(
477 path=path,
478 content=render_memory(
479 item,
480 producer=producer,
481 stale_after=stale_after_for(item, lifecycle),
482 ),
483 )
484 )
485 parent = str(Path(path).parent)
486 title = (item.text or item.id).strip().splitlines()[0][:80] or item.id
487 dirs.setdefault(parent, []).append((title, Path(path).name))
489 # Subdirectory indexes, plus links from the root index into them.
490 root_entries = list(dirs.get("", []))
491 for directory in sorted(d for d in dirs if d):
492 top = directory.split("/")[0]
493 files.append(
494 BundleFile(
495 path=f"{directory}/index.md",
496 content=render_index(dirs[directory], heading=directory, root=False),
497 )
498 )
499 if directory == top:
500 root_entries.append((top, f"{top}/"))
502 files.append(
503 BundleFile(
504 path="index.md",
505 content=render_index(root_entries, heading=f"Knowledge bundle: {bank_id}", root=True),
506 )
507 )
508 if exported:
509 files.append(BundleFile(path="log.md", content=render_log(exported)))
510 return files
513def _default_producer() -> str:
514 """``<producer>/<version>`` per SPEC 7."""
515 try:
516 import importlib.metadata as md
518 return f"astrocyte/{md.version('astrocyte')}"
519 except Exception: # pragma: no cover - version metadata is best-effort
520 return "astrocyte/unknown"
523def export_wiki_bundle(
524 pages: list[WikiPage],
525 *,
526 bank_id: str,
527 path: str | Path,
528 producer: str | None = None,
529 memories: list[VectorItem] | None = None,
530 lifecycle: Any = None,
531 allowed_roots: list[str | Path] | None = None,
532 allow_uncontained: bool = False,
533) -> ExportResult:
534 """Write an OKF bundle for ``pages`` under ``path``.
536 Path containment is enforced by :func:`astrocyte.portability._safe_resolve`
537 on every file, so a hostile ``page_id`` cannot escape the bundle root.
538 """
539 root = _safe_resolve(
540 path,
541 allowed_roots=allowed_roots,
542 allow_uncontained=allow_uncontained,
543 )
544 files = build_bundle(
545 pages,
546 bank_id=bank_id,
547 producer=producer,
548 memories=memories,
549 lifecycle=lifecycle,
550 )
552 written: list[str] = []
553 for item in files:
554 target = _safe_resolve(
555 root / item.path,
556 allowed_roots=[root],
557 allow_uncontained=False,
558 )
559 target.parent.mkdir(parents=True, exist_ok=True)
560 target.write_text(item.content, encoding="utf-8")
561 written.append(item.path)
563 concepts = [f for f in written if Path(f).name not in {"index.md", "log.md"}]
564 return ExportResult(bank_id=bank_id, concept_count=len(concepts), files=written)