"""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 ( GLOBAL_PROJECT, INBOX_PROJECT, 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('
') lines.append('

πŸ“‘ What the brain is doing

') top_actions = sorted(by_action.items(), key=lambda x: -x[1])[:6] lines.append('
' + "".join(f'{a}: {n}' for a, n in top_actions) + '
') if auto_actors: lines.append(f'

Autonomous actors: ' + " Β· ".join(f'{k} ({v})' for k, v in auto_actors.items()) + '

') lines.append('

Full timeline β†’

') lines.append('
') except Exception: pass # Issue C: Inbox + Global pseudo-projects alongside registered projects. # scope_only=True keeps real-project entities out of these counts. try: inbox_count = len(get_entities( project=INBOX_PROJECT, scope_only=True, limit=500, )) global_count = len(get_entities( project=GLOBAL_PROJECT, scope_only=True, limit=500, )) except Exception: inbox_count = 0 global_count = 0 lines.append('

πŸ“₯ Inbox & Global

') lines.append( '

Entities that don\'t belong to a specific ' 'project yet. Inbox holds pre-project leads and quotes. ' 'Global holds cross-project facts (material properties, ' 'vendor capabilities) that apply everywhere.

' ) lines.append('
') lines.append( f'' f'

πŸ“₯ Inbox

' f'

Pre-project leads, quotes, early conversations.

' f'
{inbox_count} entities
' f'
' ) lines.append( f'' f'

🌐 Global

' f'

Cross-project facts: materials, vendors, shared knowledge.

' f'
{global_count} entities
' f'
' ) lines.append('
') for bucket_name, items in buckets.items(): if not items: continue lines.append(f'

{bucket_name}

') lines.append('
') for p in items: client_line = f'
{p["client"]}
' if p["client"] else '' stage_tag = f'{p["stage"].split(" β€” ")[0]}' if p["stage"] else '' lines.append(f'') lines.append(f'

{p["id"]} {stage_tag}

') lines.append(client_line) lines.append(f'

{p["description"][:140]}

') lines.append(f'
{p["entities"]} entities Β· {p["memories"]} memories Β· {p["state"]} state
') lines.append('
') 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'
  • {s[:120]}
  • ' for s in samples[:2]) lines.append( f'
    ' f'

    {name}

    ' f'
    {count} memories
    ' f'' f'' f'
    ' ) lines.append('
    ') except Exception: pass # Quick stats all_entities = get_entities(limit=500) all_memories = get_memories(active_only=True, limit=500) pending = get_memories(status="candidate", limit=500) lines.append('

    System

    ') lines.append(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 β†’

    ' ) lines.append(f'

    Triage Queue Β· API Dashboard (JSON) Β· Health Check

    ') return render_html("AtoCore Wiki", "\n".join(lines), active_path="/wiki") def render_project(project: str) -> str: from atocore.engineering.mirror import generate_project_overview markdown_content = generate_project_overview(project) # Resolve [[Wikilinks]] before markdown so redlinks / cross-project # indicators appear in the rendered HTML. (Issue B) markdown_content = _wikilink_transform(markdown_content, current_project=project) # Convert entity names to links entities = get_entities(project=project, limit=200) html_body = md.markdown(markdown_content, extensions=["tables", "fenced_code"]) for ent in sorted(entities, key=lambda e: len(e.name), reverse=True): linked = f'{ent.name}' html_body = html_body.replace(f"{ent.name}", f"{linked}", 1) return render_html( f"{project}", html_body, breadcrumbs=[("Wiki", "/wiki"), (project, "")], ) import re as _re _WIKILINK_PATTERN = _re.compile(r"\[\[([^\[\]|]+?)(?:\|([^\[\]]+?))?\]\]") def _resolve_wikilink(target: str, current_project: str | None) -> tuple[str, str, str]: """Resolve a ``[[Name]]`` target to ``(href, css_class, extra_suffix)``. Resolution order (Issue B): 1. Same-project exact name match β†’ live link (class ``wikilink``). 2. Other-project exact name match β†’ live link with ``(in project X)`` suffix (class ``wikilink wikilink-cross``). 3. No match β†’ redlink pointing at ``/wiki/new?name=...`` so clicking opens a pre-filled "create this entity" form (class ``redlink``). """ needle = target.strip() if not needle: return ("/wiki/new", "redlink", "") same_project = None cross_project = None try: candidates = get_entities(name_contains=needle, limit=200) except Exception: candidates = [] lowered = needle.lower() for ent in candidates: if ent.name.lower() != lowered: continue if current_project and ent.project == current_project: same_project = ent break if cross_project is None: cross_project = ent if same_project is not None: return (f"/wiki/entities/{same_project.id}", "wikilink", "") if cross_project is not None: suffix = ( f' (in {cross_project.project})' if cross_project.project else ' (global)' ) return (f"/wiki/entities/{cross_project.id}", "wikilink wikilink-cross", suffix) from urllib.parse import quote href = f"/wiki/new?name={quote(needle)}" if current_project: href += f"&project={quote(current_project)}" return (href, "redlink", "") def _wikilink_transform(text: str, current_project: str | None) -> str: """Replace ``[[Name]]`` / ``[[Name|Display]]`` tokens with HTML anchors. Runs before markdown rendering. Emits raw HTML which python-markdown preserves unchanged. """ if not text or "[[" not in text: return text def _sub(match: _re.Match) -> str: target = match.group(1) display = (match.group(2) or target).strip() href, cls, suffix = _resolve_wikilink(target, current_project) title = "create this entity" if cls == "redlink" else target.strip() return ( f'' f'{_escape_html(display)}{suffix}' ) return _WIKILINK_PATTERN.sub(_sub, text) def render_new_entity_form(name: str = "", project: str = "") -> str: """Issue B β€” "create this entity" form targeted by redlinks.""" from atocore.engineering.service import ENTITY_TYPES safe_name = _escape_attr(name or "") safe_project = _escape_attr(project or "") opts = "".join( f'' for t in ENTITY_TYPES ) lines = [ '

    Create entity

    ', ('

    This entity was referenced via a wikilink but does not exist yet. ' 'Fill in the details to create it β€” the wiki link will resolve on reload.

    '), '
    ', f'', f'', f'', '', '', '
    ', '
    ', """""", ] return render_html( f"Create {name}" if name else "Create entity", "\n".join(lines), breadcrumbs=[("Wiki", "/wiki"), ("Create entity", "")], ) def _render_visual_evidence(entity_id: str, ctx: dict) -> str: """Render EVIDENCED_BY β†’ artifact links as an inline thumbnail strip.""" from atocore.assets import get_asset artifacts = [] for rel in ctx["relationships"]: if rel.source_entity_id != entity_id or rel.relationship_type != "evidenced_by": continue target = ctx["related_entities"].get(rel.target_entity_id) if target is None or target.entity_type != "artifact": continue artifacts.append(target) if not artifacts: return "" tiles = [] for art in artifacts: props = art.properties or {} kind = props.get("kind", "other") caption = props.get("caption", art.name) asset_id = props.get("asset_id") asset = get_asset(asset_id) if asset_id else None detail_href = f"/wiki/entities/{art.id}" if kind == "image" and asset and asset.mime_type.startswith("image/"): full_href = f"/assets/{asset.id}" thumb = f"/assets/{asset.id}/thumbnail?size=240" tiles.append( f'
    ' f'' f'{_escape_attr(caption)}' f'' f'
    {_escape_html(caption)}
    ' f'
    ' ) elif kind == "pdf" and asset: full_href = f"/assets/{asset.id}" tiles.append( f'
    ' f'' f'πŸ“„ PDF: {_escape_html(caption)}' f' Β· details' f'
    ' ) else: tiles.append( f'
    ' f'πŸ“Ž {_escape_html(caption)}' f' {kind}' f'
    ' ) return ( '

    Visual evidence

    ' f'
    {"".join(tiles)}
    ' ) def _render_artifact_body(ent) -> list[str]: """Render an artifact entity's own image/pdf/caption.""" from atocore.assets import get_asset props = ent.properties or {} kind = props.get("kind", "other") caption = props.get("caption", "") capture_context = props.get("capture_context", "") asset_id = props.get("asset_id") asset = get_asset(asset_id) if asset_id else None out: list[str] = [] if kind == "image" and asset and asset.mime_type.startswith("image/"): out.append( f'
    ' f'' f'' f'' f'
    {_escape_html(caption)}
    ' f'
    ' ) elif kind == "pdf" and asset: out.append( f'

    πŸ“„ ' f'Open PDF ({asset.size_bytes // 1024} KB)

    ' ) elif asset_id: out.append(f'

    asset_id: {asset_id} β€” blob missing

    ') if capture_context: out.append('

    Capture context

    ') out.append(f'
    {_escape_html(capture_context)}
    ') return out def _escape_html(s: str) -> str: if s is None: return "" return (str(s) .replace("&", "&") .replace("<", "<") .replace(">", ">")) def _escape_attr(s: str) -> str: return _escape_html(s).replace('"', """) def render_entity(entity_id: str) -> str | None: ctx = get_entity_with_context(entity_id) if ctx is None: return None ent = ctx["entity"] lines = [f'

    [{ent.entity_type}] {ent.name}

    '] if ent.project: lines.append(f'

    Project: {ent.project}

    ') if ent.description: desc_html = _wikilink_transform(ent.description, current_project=ent.project) lines.append(f'

    {desc_html}

    ') if ent.properties: lines.append('

    Properties

    ') lines.append(f'

    confidence: {ent.confidence} Β· status: {ent.status} Β· created: {ent.created_at}

    ') # Issue F: artifact entities render their own image inline; other # entities render their EVIDENCED_BY artifacts as a visual strip. if ent.entity_type == "artifact": lines.extend(_render_artifact_body(ent)) else: evidence_html = _render_visual_evidence(ent.id, ctx) if evidence_html: lines.append(evidence_html) if ctx["relationships"]: lines.append('

    Relationships

    ') breadcrumbs = [("Wiki", "/wiki")] if ent.project: breadcrumbs.append((ent.project, f"/wiki/projects/{ent.project}")) breadcrumbs.append((ent.name, "")) return render_html(ent.name, "\n".join(lines), breadcrumbs=breadcrumbs) def render_search(query: str) -> str: lines = [f'

    Search: "{query}"

    '] # Search entities by name entities = get_entities(name_contains=query, limit=20) if entities: lines.append(f'

    Entities ({len(entities)})

    ') # 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)})

    ') 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.

    ') lines.append("""
    """) lines.append( '

    How this works

    ' '' '

    The extractor is aggressive about capturing signal β€” don\'t hand-filter. ' 'If the conversation had nothing durable, triage will auto-reject.

    ' ) return render_html( "Capture β€” AtoCore", "\n".join(lines), breadcrumbs=[("Wiki", "/wiki"), ("Capture", "")], active_path="/wiki/capture", ) # --------------------------------------------------------------------- # Phase 7E β€” /wiki/memories/{id}: memory detail page # --------------------------------------------------------------------- def render_memory_detail(memory_id: str) -> str | None: """Full view of a single memory: content, audit trail, source refs, neighbors, graduation status. Fills the drill-down gap the list views can't.""" from atocore.memory.service import get_memory_audit from atocore.models.database import get_connection with get_connection() as conn: row = conn.execute("SELECT * FROM memories WHERE id = ?", (memory_id,)).fetchone() if row is None: return None import json as _json mem = dict(row) try: tags = _json.loads(mem.get("domain_tags") or "[]") or [] except Exception: tags = [] lines = [f'

    {mem["memory_type"]}: {mem["content"][:80]}

    '] if len(mem["content"]) > 80: lines.append(f'

    {mem["content"]}

    ') # Metadata row meta_items = [ f'{mem["status"]}', f'{mem["memory_type"]}', ] if mem.get("project"): meta_items.append(f'{mem["project"]}') meta_items.append(f'confidence: {float(mem.get("confidence") or 0):.2f}') meta_items.append(f'refs: {int(mem.get("reference_count") or 0)}') if mem.get("valid_until"): meta_items.append(f'valid until {str(mem["valid_until"])[:10]}') lines.append(f'

    {" Β· ".join(meta_items)}

    ') if tags: tag_links = " ".join(f'{t}' for t in tags) lines.append(f'

    {tag_links}

    ') lines.append(f'

    id: {mem["id"]} Β· created: {mem["created_at"]}' f' Β· updated: {mem.get("updated_at", "?")}' + (f' Β· last referenced: {mem["last_referenced_at"]}' if mem.get("last_referenced_at") else '') + '

    ') # Graduation if mem.get("graduated_to_entity_id"): eid = mem["graduated_to_entity_id"] lines.append( f'

    πŸŽ“ Graduated

    ' f'

    This memory was promoted to a typed entity: ' f'{eid[:8]}

    ' ) # Source chunk if mem.get("source_chunk_id"): lines.append(f'

    Source chunk

    {mem["source_chunk_id"]}

    ') # Audit trail audit = get_memory_audit(memory_id, limit=50) if audit: lines.append(f'

    Audit trail ({len(audit)} events)

    ') # 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'

    Related (by tag)

    ') 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)})

    ') # 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'

    πŸ”§ Entities ({len(ent_matching)})

    ') except Exception: pass return render_html( f"Domain: {tag}", "\n".join(lines), breadcrumbs=[("Wiki", "/wiki"), ("Domains", ""), (tag, "")], ) # --------------------------------------------------------------------- # /wiki/activity β€” autonomous-activity feed # --------------------------------------------------------------------- def render_activity(hours: int = 48, limit: int = 100) -> str: """Timeline of what the autonomous pipeline did recently. Answers 'what has the brain been doing while I was away?'""" from atocore.memory.service import get_recent_audit audit = get_recent_audit(limit=limit) # Group events by category for summary by_action: dict[str, int] = {} by_actor: dict[str, int] = {} for a in audit: by_action[a["action"]] = by_action.get(a["action"], 0) + 1 by_actor[a["actor"]] = by_actor.get(a["actor"], 0) + 1 lines = [f'

    πŸ“‘ Activity Feed

    '] lines.append(f'

    Last {len(audit)} events in the memory audit log

    ') # Summary chips if by_action or by_actor: lines.append('

    Summary

    ') lines.append('

    By action: ' + " Β· ".join(f'{k}: {v}' for k, v in sorted(by_action.items(), key=lambda x: -x[1])) + '

    ') lines.append('

    By actor: ' + " Β· ".join(f'{k}: {v}' for k, v in sorted(by_actor.items(), key=lambda x: -x[1])) + '

    ') # Action-type color/emoji action_emoji = { "created": "βž•", "promoted": "βœ…", "rejected": "❌", "invalidated": "🚫", "superseded": "πŸ”€", "reinforced": "πŸ”", "updated": "✏️", "auto_promoted": "⚑", "created_via_merge": "πŸ”—", "valid_until_extended": "⏳", "tag_canonicalized": "🏷️", } lines.append('

    Timeline

    ') return render_html( "Activity β€” AtoCore", "\n".join(lines), breadcrumbs=[("Wiki", "/wiki"), ("Activity", "")], active_path="/wiki/activity", ) _TEMPLATE = """ {{title}} β€” AtoCore {{nav}} {{body}} """