"""AtoCore Wiki β navigable HTML pages from structured data.
A lightweight wiki served directly from the AtoCore API. Every page is
generated on-demand from the database so it's always current. Source of
truth is the database β the wiki is a derived view.
Routes:
/wiki Homepage with project list + search
/wiki/projects/{name} Full project overview
/wiki/entities/{id} Entity detail with relationships
/wiki/search?q=... Search entities, memories, state
"""
from __future__ import annotations
import markdown as md
from atocore.context.project_state import get_state
from atocore.engineering.service import (
get_entities,
get_entity,
get_entity_with_context,
get_relationships,
)
from atocore.memory.service import get_memories
from atocore.projects.registry import load_project_registry
_TOP_NAV_LINKS = [
("π Home", "/wiki"),
("π‘ Activity", "/wiki/activity"),
("π Triage", "/admin/triage"),
("π Dashboard", "/admin/dashboard"),
]
def _render_topnav(active_path: str = "") -> str:
items = []
for label, href in _TOP_NAV_LINKS:
cls = "topnav-item active" if href == active_path else "topnav-item"
items.append(f'{label}')
return f''
def render_html(title: str, body_html: str, breadcrumbs: list[tuple[str, str]] | None = None, active_path: str = "") -> str:
topnav = _render_topnav(active_path)
crumbs = ""
if breadcrumbs:
parts = []
for label, href in breadcrumbs:
if href:
parts.append(f'{label}')
else:
parts.append(f"{label}")
crumbs = f''
nav = topnav + crumbs
return _TEMPLATE.replace("{{title}}", title).replace("{{nav}}", nav).replace("{{body}}", body_html)
def render_homepage() -> str:
projects = []
try:
registered = load_project_registry()
for p in registered:
entity_count = len(get_entities(project=p.project_id, limit=200))
memory_count = len(get_memories(project=p.project_id, active_only=True, limit=200))
state_entries = get_state(p.project_id)
# Pull stage/type/client from state entries
stage = ""
proj_type = ""
client = ""
for e in state_entries:
if e.category == "status":
if e.key == "stage":
stage = e.value
elif e.key == "type":
proj_type = e.value
elif e.key == "client":
client = e.value
projects.append({
"id": p.project_id,
"description": p.description,
"entities": entity_count,
"memories": memory_count,
"state": len(state_entries),
"stage": stage,
"type": proj_type,
"client": client,
})
except Exception:
pass
# Group by high-level bucket
buckets: dict[str, list] = {
"Active Contracts": [],
"Leads & Prospects": [],
"Internal Tools & Infra": [],
"Other": [],
}
for p in projects:
t = p["type"].lower()
s = p["stage"].lower()
if "lead" in t or "lead" in s or "prospect" in s:
buckets["Leads & Prospects"].append(p)
elif "contract" in t or ("active" in s and "contract" in s):
buckets["Active Contracts"].append(p)
elif "infra" in t or "tool" in t or "internal" in t:
buckets["Internal Tools & Infra"].append(p)
else:
buckets["Other"].append(p)
lines = ['
AtoCore Wiki
']
lines.append('')
# What's happening β autonomous activity snippet
try:
from atocore.memory.service import get_recent_audit
recent = get_recent_audit(limit=30)
by_action: dict[str, int] = {}
by_actor: dict[str, int] = {}
for a in recent:
by_action[a["action"]] = by_action.get(a["action"], 0) + 1
by_actor[a["actor"]] = by_actor.get(a["actor"], 0) + 1
# Surface autonomous actors specifically
auto_actors = {k: v for k, v in by_actor.items()
if k.startswith("auto-") or k == "confidence-decay"
or k == "phase10-auto-promote" or k == "transient-to-durable"}
if recent:
lines.append('
')
# Phase 6 C.2: Emerging projects section
try:
import json as _json
emerging_projects = []
state_entries = get_state("atocore")
for e in state_entries:
if e.category == "proposals" and e.key == "unregistered_projects":
try:
emerging_projects = _json.loads(e.value)
except Exception:
emerging_projects = []
break
if emerging_projects:
lines.append('
π Emerging
')
lines.append('
Projects that appear in memories but aren\'t yet registered. '
'One click to promote them to first-class projects.
')
lines.append('
')
for ep in emerging_projects[:10]:
name = ep.get("project", "?")
count = ep.get("count", 0)
samples = ep.get("sample_contents", [])
samples_html = "".join(f'
{len(all_entities)} entities Β· {len(all_memories)} active memories Β· {len(projects)} projects
')
# Triage queue prompt β surfaced prominently if non-empty
if pending:
tone = "triage-warning" if len(pending) > 50 else "triage-notice"
lines.append(
f'
ποΈ {len(pending)} candidates awaiting triage β '
f'review now β
')
for rel in ctx["relationships"]:
other_id = rel.target_entity_id if rel.source_entity_id == entity_id else rel.source_entity_id
other = ctx["related_entities"].get(other_id)
if other:
direction = "\u2192" if rel.source_entity_id == entity_id else "\u2190"
lines.append(
f'
')
# Search memories β match on content OR domain_tags (Phase 3)
all_memories = get_memories(active_only=True, limit=200)
query_lower = query.lower()
matching_mems = [
m for m in all_memories
if query_lower in m.content.lower()
or any(query_lower in (t or "").lower() for t in (m.domain_tags or []))
][:20]
if matching_mems:
lines.append(f'
Memories ({len(matching_mems)})
')
for m in matching_mems:
proj = f' {m.project}' if m.project else ''
tags_html = ""
if m.domain_tags:
tag_links = " ".join(
f'{t}'
for t in m.domain_tags[:5]
)
tags_html = f' {tag_links}'
expiry_html = ""
if m.valid_until:
expiry_html = f' valid until {m.valid_until[:10]}'
lines.append(
f'
')
if not entities and not matching_mems:
lines.append('
No results found.
')
lines.append('')
return render_html(
f"Search: {query}",
"\n".join(lines),
breadcrumbs=[("Wiki", "/wiki"), ("Search", "")],
)
# ---------------------------------------------------------------------
# /wiki/capture β DEPRECATED emergency paste-in form.
# Kept as an endpoint because POST /interactions is public anyway, but
# REMOVED from the topnav so it's not promoted as the capture path.
# The sanctioned surfaces are Claude Code (Stop + UserPromptSubmit
# hooks) and OpenClaw (capture plugin with 7I context injection).
# This form is explicitly a last-resort for when someone has to feed
# in an external log and can't get the normal hooks to reach it.
# ---------------------------------------------------------------------
def render_capture() -> str:
lines = ['
π₯ Manual capture (fallback only)
']
lines.append(
'
This is not the capture path. '
'The sanctioned capture surfaces are Claude Code (Stop hook auto-captures every turn) '
'and OpenClaw (plugin auto-captures + injects AtoCore context on every agent turn). '
'This form exists only as a last resort for external logs you can\'t get into the normal pipeline.
'
)
lines.append(
'
If you\'re reaching for this page because you had a chat somewhere AtoCore didn\'t see, '
'fix the capture surface instead β don\'t paste. The deliberate scope is Claude Code + OpenClaw.
'
)
lines.append('
Your prompt + the assistant\'s response. Project is optional β '
'the extractor infers it from content.
')
# Neighbors by shared tag
if tags:
from atocore.memory.service import get_memories as _get_memories
neighbors = []
for t in tags[:3]:
for other in _get_memories(active_only=True, limit=30):
if other.id == memory_id:
continue
if any(ot == t for ot in (other.domain_tags or [])):
neighbors.append(other)
# Dedupe
seen = set()
uniq = []
for n in neighbors:
if n.id in seen:
continue
seen.add(n.id)
uniq.append(n)
if uniq:
lines.append(f'
')
return render_html(
f"Memory {memory_id[:8]}",
"\n".join(lines),
breadcrumbs=[("Wiki", "/wiki"), ("Memory", "")],
)
# ---------------------------------------------------------------------
# Phase 7F β /wiki/domains/{tag}: cross-project domain view
# ---------------------------------------------------------------------
def render_domain(tag: str) -> str:
"""All memories + entities carrying a given domain_tag, grouped by project.
Answers 'what does the brain know about optics, across all projects?'"""
tag = (tag or "").strip().lower()
if not tag:
return render_html("Domain", "
No tag specified.
",
breadcrumbs=[("Wiki", "/wiki"), ("Domains", "")])
all_mems = get_memories(active_only=True, limit=500)
matching = [m for m in all_mems
if any((t or "").lower() == tag for t in (m.domain_tags or []))]
# Group by project
by_project: dict[str, list] = {}
for m in matching:
by_project.setdefault(m.project or "(global)", []).append(m)
lines = [f'
Domain: {tag}
']
lines.append(f'
{len(matching)} active memories across {len(by_project)} projects
')
if not matching:
lines.append(
f'
No memories currently carry the tag {tag}.
'
'
Domain tags are assigned by the extractor when it identifies '
'the topical scope of a memory. They update over time.
'
)
return render_html(
f"Domain: {tag}",
"\n".join(lines),
breadcrumbs=[("Wiki", "/wiki"), ("Domains", ""), (tag, "")],
)
# Sort projects by count descending, (global) last
def sort_key(item: tuple[str, list]) -> tuple[int, int]:
proj, mems = item
return (1 if proj == "(global)" else 0, -len(mems))
for proj, mems in sorted(by_project.items(), key=sort_key):
proj_link = proj if proj == "(global)" else f'{proj}'
lines.append(f'
{proj_link} ({len(mems)})
')
for m in mems:
other_tags = [t for t in (m.domain_tags or []) if t != tag][:3]
other_tags_html = ""
if other_tags:
other_tags_html = ' ' + " ".join(
f'{t}' for t in other_tags
) + ''
lines.append(
f'
')
# Entities with this tag (if any have tags β currently they might not)
try:
all_entities = get_entities(limit=500)
ent_matching = []
for e in all_entities:
tags = e.properties.get("domain_tags") if e.properties else []
if isinstance(tags, list) and tag in [str(t).lower() for t in tags]:
ent_matching.append(e)
if ent_matching:
lines.append(f'
')
for a in audit:
emoji = action_emoji.get(a["action"], "β’")
preview = a.get("content_preview") or ""
ts_short = a["timestamp"][:16] if a.get("timestamp") else "?"
mid_short = (a.get("memory_id") or "")[:8]
note = f' β {a["note"]}' if a.get("note") else ""
lines.append(
f'