WorldMM Live Memory System — Implementation Plan

main/server/worldmm/
├── __init__.py
├── ingest/
│   ├── __init__.py
│   ├── captioner.py           # build_caption_prompt(), parse_caption_response()
│   └── transcriber.py         # transcribe_audio() — Whisper API client
├── memory/
│   ├── __init__.py
│   ├── episodic/
│   │   ├── __init__.py
│   │   ├── graph.py           # EpisodicGraph: build igraph from triples, run PPR
│   │   ├── openie.py          # build_ner_prompt(), parse_ner(), build_triple_prompt(), parse_triples()
│   │   └── multiscale.py      # build_merge_prompt(), parse_merge_response()
│   ├── semantic/
│   │   ├── __init__.py
│   │   ├── graph.py           # SemanticGraph: single evolving entity graph, PPR with edge scoring
│   │   ├── extraction.py      # build_semantic_triple_prompt(), parse_semantic_triples()
│   │   └── consolidation.py   # find_candidates(), build_consolidation_prompt(), apply_consolidation()
│   └── visual/
│       ├── __init__.py
│       ├── encoder.py         # VLM2VecClient: HTTP calls to GPU worker
│       └── index.py           # query_visual_embeddings() — pgvector cosine similarity
├── retrieval/
│   ├── __init__.py
│   ├── agent.py               # ReasoningAgent: SEARCH/ANSWER loop, up to 5 rounds
│   ├── episodic_retriever.py  # retrieve_episodic(): PPR across scales + cross-scale rerank
│   ├── semantic_retriever.py  # retrieve_semantic(): PPR on entity graph
│   └── visual_retriever.py    # retrieve_visual(): VLM2Vec text query → cosine sim
├── llm/
│   ├── __init__.py
│   ├── client.py              # LLMClient: build_request(), call(), supports batch/direct
│   └── batch.py               # BatchClient: submit(), poll(), collect_results()
├── llm/prompts/
│   ├── caption.txt
│   ├── ner.txt
│   ├── episodic_triples.txt
│   ├── semantic_triples.txt
│   ├── merge_captions.txt
│   ├── consolidation.txt
│   ├── cross_scale_rerank.txt
│   ├── reasoning_agent.txt
│   ├── response.txt
│   └── entity_confirm.txt
└── tests/
    ├── __init__.py
    ├── conftest.py            # shared fixtures, mock LLM helper
    ├── fixtures/              # EgoLife-derived canned data
    │   ├── sample_caption.txt
    │   ├── sample_ner_response.json
    │   ├── sample_triples_response.json
    │   └── sample_semantic_triples_response.json
    ├── test_llm_client.py
    ├── test_openie.py
    ├── test_entity_resolution.py
    ├── test_captioner.py
    ├── test_multiscale.py
    ├── test_episodic_graph.py
    ├── test_semantic_graph.py
    ├── test_consolidation.py
    ├── test_visual_encoder.py
    ├── test_episodic_retriever.py
    ├── test_semantic_retriever.py
    ├── test_visual_retriever.py
    └── test_retrieval_agent.py

ORM models (shared layer):
├── main/server/layers/shared/python/shared/orm/worldmm_orm.py

Entity resolution:
├── main/server/worldmm/memory/episodic/entity_resolution.py  # resolve_entities(): embed → HNSW → LLM confirm → create/alias

# main/server/worldmm/tests/test_orm.py
from __future__ import annotations

import sys
from pathlib import Path
from uuid import uuid4

import pytest

_server_root = Path(__file__).resolve().parents[3]
_layer_root = _server_root / "layers" / "shared" / "python"
for p in (_layer_root, _server_root):
    if str(p) not in sys.path:
        sys.path.insert(0, str(p))


def _setup_db():
    import os
    os.environ.setdefault("DATABASE_URL", "sqlite:///:memory:")
    import shared.orm.orm as orm
    orm.getEngine.cache_clear()
    orm.getSessionFactory.cache_clear()
    # Must import worldmm_orm so models are registered on Base
    import shared.orm.worldmm_orm  # noqa: F401
    orm.initDb()


def test_create_segment():
    _setup_db()
    from shared.orm.worldmm_orm import WorldMMSegment, create_segment
    segment_id = create_segment(
        user_id="user-1",
        start_time="2026-03-22T10:00:00Z",
        end_time="2026-03-22T10:00:30Z",
        duration_seconds=30,
        caption="I stand at the dining table",
        s3_video_key="worldmm/user-1/clips/001.mp4",
        transcript="I'm standing at the dining table",
    )
    assert segment_id is not None
    assert len(segment_id) == 36  # UUID format


def test_create_entity_canonical():
    _setup_db()
    from shared.orm.worldmm_orm import create_entity
    entity_id = create_entity(
        user_id="user-1",
        surface_form="Katrina",
        canonical_name="Katrina",
        canonical_entity_id=None,
    )
    assert entity_id is not None


def test_create_entity_alias():
    _setup_db()
    from shared.orm.worldmm_orm import create_entity
    canonical_id = create_entity(
        user_id="user-1",
        surface_form="Katrina",
        canonical_name="Katrina",
        canonical_entity_id=None,
    )
    alias_id = create_entity(
        user_id="user-1",
        surface_form="Kate",
        canonical_name=None,
        canonical_entity_id=canonical_id,
    )
    assert alias_id != canonical_id


def test_create_triple_with_entity_object():
    _setup_db()
    from shared.orm.worldmm_orm import create_entity, create_segment, create_triple
    seg_id = create_segment(
        user_id="user-1",
        start_time="2026-03-22T10:00:00Z",
        end_time="2026-03-22T10:00:30Z",
        duration_seconds=30,
        caption="test",
    )
    subj_id = create_entity(user_id="user-1", surface_form="I", canonical_name="I")
    obj_id = create_entity(user_id="user-1", surface_form="Katrina", canonical_name="Katrina")
    triple_id = create_triple(
        segment_id=seg_id,
        user_id="user-1",
        memory_type="episodic",
        subject_entity_id=subj_id,
        predicate="talks to",
        object_entity_id=obj_id,
    )
    assert triple_id is not None


def test_create_triple_with_literal_object():
    _setup_db()
    from shared.orm.worldmm_orm import create_entity, create_segment, create_triple
    seg_id = create_segment(
        user_id="user-1",
        start_time="2026-03-22T10:00:00Z",
        end_time="2026-03-22T10:00:30Z",
        duration_seconds=30,
        caption="test",
    )
    subj_id = create_entity(user_id="user-1", surface_form="meeting", canonical_name="meeting")
    triple_id = create_triple(
        segment_id=seg_id,
        user_id="user-1",
        memory_type="episodic",
        subject_entity_id=subj_id,
        predicate="started_at",
        object_literal="2:30 PM",
    )
    assert triple_id is not None


def test_create_triple_rejects_both_object_types():
    _setup_db()
    from shared.orm.worldmm_orm import create_entity, create_segment, create_triple
    seg_id = create_segment(
        user_id="user-1",
        start_time="2026-03-22T10:00:00Z",
        end_time="2026-03-22T10:00:30Z",
        duration_seconds=30,
        caption="test",
    )
    subj_id = create_entity(user_id="user-1", surface_form="I", canonical_name="I")
    obj_id = create_entity(user_id="user-1", surface_form="Katrina", canonical_name="Katrina")
    with pytest.raises(ValueError, match="exactly one"):
        create_triple(
            segment_id=seg_id,
            user_id="user-1",
            memory_type="episodic",
            subject_entity_id=subj_id,
            predicate="talks to",
            object_entity_id=obj_id,
            object_literal="some text",
        )


def test_get_triples_for_user():
    _setup_db()
    from shared.orm.worldmm_orm import (
        create_entity, create_segment, create_triple, get_triples_for_user,
    )
    seg_id = create_segment(
        user_id="user-1",
        start_time="2026-03-22T10:00:00Z",
        end_time="2026-03-22T10:00:30Z",
        duration_seconds=30,
        caption="test",
    )
    subj_id = create_entity(user_id="user-1", surface_form="I", canonical_name="I")
    obj_id = create_entity(user_id="user-1", surface_form="Katrina", canonical_name="Katrina")
    create_triple(
        segment_id=seg_id, user_id="user-1", memory_type="episodic",
        subject_entity_id=subj_id, predicate="talks to", object_entity_id=obj_id,
    )
    create_triple(
        segment_id=seg_id, user_id="user-1", memory_type="semantic",
        subject_entity_id=subj_id, predicate="is friends with", object_entity_id=obj_id,
    )
    episodic = get_triples_for_user("user-1", "episodic")
    semantic = get_triples_for_user("user-1", "semantic")
    assert len(episodic) == 1
    assert len(semantic) == 1
    assert episodic[0]["predicate"] == "talks to"


def test_invalidate_triple():
    _setup_db()
    from shared.orm.worldmm_orm import (
        create_entity, create_segment, create_triple,
        invalidate_triple, get_triples_for_user,
    )
    seg_id = create_segment(
        user_id="user-1",
        start_time="2026-03-22T10:00:00Z",
        end_time="2026-03-22T10:00:30Z",
        duration_seconds=30,
        caption="test",
    )
    subj_id = create_entity(user_id="user-1", surface_form="I", canonical_name="I")
    obj_id = create_entity(user_id="user-1", surface_form="Katrina", canonical_name="Katrina")
    triple_id = create_triple(
        segment_id=seg_id, user_id="user-1", memory_type="semantic",
        subject_entity_id=subj_id, predicate="is friends with", object_entity_id=obj_id,
    )
    invalidate_triple(triple_id)
    active = get_triples_for_user("user-1", "semantic")
    assert len(active) == 0

# main/server/worldmm/__init__.py
# (empty)

# main/server/worldmm/tests/__init__.py
# (empty)

# main/server/worldmm/tests/conftest.py
# (empty for now — fixtures added in later tasks)

# main/server/layers/shared/python/shared/orm/worldmm_orm.py
from __future__ import annotations

import uuid
from datetime import datetime, timezone
from typing import Any

from sqlalchemy import Column, Integer, String, Text, DateTime, Enum as SAEnum, ForeignKey, CheckConstraint
from sqlalchemy.orm import relationship

from .orm import Base, getSession


class WorldMMSegment(Base):
    __tablename__ = "worldmm_segments"

    id: str = Column(String, primary_key=True)
    user_id: str = Column(String, nullable=False)
    start_time: str = Column(String, nullable=False)
    end_time: str = Column(String, nullable=False)
    duration_seconds: int = Column(Integer, nullable=False)
    caption: str = Column(Text, nullable=True)
    s3_video_key: str = Column(String, nullable=True)
    s3_frames_key: str = Column(String, nullable=True)
    transcript: str = Column(Text, nullable=True)
    parent_segment_id: str = Column(String, ForeignKey("worldmm_segments.id"), nullable=True)


class WorldMMEntity(Base):
    __tablename__ = "worldmm_entities"

    id: str = Column(String, primary_key=True)
    user_id: str = Column(String, nullable=False)
    canonical_entity_id: str = Column(String, ForeignKey("worldmm_entities.id"), nullable=True)
    surface_form: str = Column(String, nullable=False)
    canonical_name: str = Column(String, nullable=True)
    # Note: embedding column (VECTOR(3072)) for OpenAI text-embedding-3-large
    # Omitted from SQLite tests; added via Alembic migration for PostgreSQL
    # Entity resolution tests mock the embedding lookup directly


class WorldMMTriple(Base):
    __tablename__ = "worldmm_triples"
    __table_args__ = (
        CheckConstraint(
            "(object_entity_id IS NOT NULL AND object_literal IS NULL) OR "
            "(object_entity_id IS NULL AND object_literal IS NOT NULL)",
            name="ck_triple_object_xor",
        ),
    )

    id: str = Column(String, primary_key=True)
    segment_id: str = Column(String, ForeignKey("worldmm_segments.id"), nullable=False)
    user_id: str = Column(String, nullable=False)
    memory_type: str = Column(String, nullable=False)  # "episodic" or "semantic"
    subject_entity_id: str = Column(String, ForeignKey("worldmm_entities.id"), nullable=False)
    predicate: str = Column(String, nullable=False)
    object_entity_id: str = Column(String, ForeignKey("worldmm_entities.id"), nullable=True)
    object_literal: str = Column(String, nullable=True)
    created_at: str = Column(String, nullable=False)
    invalidated_at: str = Column(String, nullable=True)


class WorldMMVisualEmbedding(Base):
    __tablename__ = "worldmm_visual_embeddings"

    id: str = Column(String, primary_key=True)
    segment_id: str = Column(String, ForeignKey("worldmm_segments.id"), nullable=False)
    user_id: str = Column(String, nullable=False)
    timestamp: str = Column(String, nullable=False)
    # Note: embedding column (Vector) added only when using PostgreSQL + pgvector
    # SQLite tests skip vector operations


def create_segment(
    user_id: str,
    start_time: str,
    end_time: str,
    duration_seconds: int,
    caption: str | None = None,
    s3_video_key: str | None = None,
    s3_frames_key: str | None = None,
    transcript: str | None = None,
    parent_segment_id: str | None = None,
) -> str:
    segment_id = str(uuid.uuid4())
    session = getSession()
    try:
        session.add(WorldMMSegment(
            id=segment_id,
            user_id=user_id,
            start_time=start_time,
            end_time=end_time,
            duration_seconds=duration_seconds,
            caption=caption,
            s3_video_key=s3_video_key,
            s3_frames_key=s3_frames_key,
            transcript=transcript,
            parent_segment_id=parent_segment_id,
        ))
        session.commit()
        return segment_id
    finally:
        session.close()


def create_entity(
    user_id: str,
    surface_form: str,
    canonical_name: str | None = None,
    canonical_entity_id: str | None = None,
) -> str:
    entity_id = str(uuid.uuid4())
    session = getSession()
    try:
        session.add(WorldMMEntity(
            id=entity_id,
            user_id=user_id,
            canonical_entity_id=canonical_entity_id,
            surface_form=surface_form,
            canonical_name=canonical_name,
        ))
        session.commit()
        return entity_id
    finally:
        session.close()


def create_triple(
    segment_id: str,
    user_id: str,
    memory_type: str,
    subject_entity_id: str,
    predicate: str,
    object_entity_id: str | None = None,
    object_literal: str | None = None,
) -> str:
    if (object_entity_id is None) == (object_literal is None):
        raise ValueError("Triple must have exactly one of object_entity_id or object_literal")
    triple_id = str(uuid.uuid4())
    now = datetime.now(timezone.utc).isoformat()
    session = getSession()
    try:
        session.add(WorldMMTriple(
            id=triple_id,
            segment_id=segment_id,
            user_id=user_id,
            memory_type=memory_type,
            subject_entity_id=subject_entity_id,
            predicate=predicate,
            object_entity_id=object_entity_id,
            object_literal=object_literal,
            created_at=now,
            invalidated_at=None,
        ))
        session.commit()
        return triple_id
    finally:
        session.close()


def get_triples_for_user(
    user_id: str,
    memory_type: str,
) -> list[dict[str, Any]]:
    session = getSession()
    try:
        rows = (
            session.query(WorldMMTriple)
            .filter(
                WorldMMTriple.user_id == user_id,
                WorldMMTriple.memory_type == memory_type,
                WorldMMTriple.invalidated_at.is_(None),
            )
            .all()
        )
        return [
            {
                "id": r.id,
                "segment_id": r.segment_id,
                "subject_entity_id": r.subject_entity_id,
                "predicate": r.predicate,
                "object_entity_id": r.object_entity_id,
                "object_literal": r.object_literal,
            }
            for r in rows
        ]
    finally:
        session.close()


def invalidate_triple(triple_id: str) -> None:
    now = datetime.now(timezone.utc).isoformat()
    session = getSession()
    try:
        record = session.get(WorldMMTriple, triple_id)
        if record is None:
            raise ValueError("Triple not found")
        record.invalidated_at = now
        session.commit()
    finally:
        session.close()

git add main/server/layers/shared/python/shared/orm/worldmm_orm.py \
        main/server/worldmm/__init__.py \
        main/server/worldmm/tests/__init__.py \
        main/server/worldmm/tests/conftest.py \
        main/server/worldmm/tests/test_orm.py
git commit -m "feat(worldmm): add ORM models for segments, entities, triples, visual embeddings"

# main/server/worldmm/tests/test_llm_client.py
from __future__ import annotations

import sys
from pathlib import Path
from unittest.mock import MagicMock, patch

_server_root = Path(__file__).resolve().parents[3]
_layer_root = _server_root / "layers" / "shared" / "python"
for p in (_layer_root, _server_root):
    if str(p) not in sys.path:
        sys.path.insert(0, str(p))


def test_build_request_includes_reasoning_effort():
    from worldmm.llm.client import build_request
    req = build_request(
        model="gpt-5-mini",
        messages=[{"role": "user", "content": "hello"}],
        max_completion_tokens=256,
        reasoning_effort="low",
    )
    assert req["model"] == "gpt-5-mini"
    assert req["max_completion_tokens"] == 256
    assert req["reasoning_effort"] == "low"
    assert req["messages"] == [{"role": "user", "content": "hello"}]


def test_build_request_without_reasoning_effort():
    from worldmm.llm.client import build_request
    req = build_request(
        model="gpt-5",
        messages=[{"role": "user", "content": "hello"}],
        max_completion_tokens=1024,
    )
    assert "reasoning_effort" not in req


def test_call_direct_mode():
    from worldmm.llm.client import LLMClient
    mock_openai = MagicMock()
    mock_openai.chat.completions.create.return_value = MagicMock(
        choices=[MagicMock(message=MagicMock(content="test response"))]
    )
    client = LLMClient(mode="direct", openai_client=mock_openai)
    result = client.call(build_request_dict={
        "model": "gpt-5-mini",
        "messages": [{"role": "user", "content": "hi"}],
        "max_completion_tokens": 256,
        "reasoning_effort": "low",
    })
    assert result == "test response"
    mock_openai.chat.completions.create.assert_called_once()


def test_call_direct_passes_all_params():
    from worldmm.llm.client import LLMClient
    mock_openai = MagicMock()
    mock_openai.chat.completions.create.return_value = MagicMock(
        choices=[MagicMock(message=MagicMock(content="ok"))]
    )
    client = LLMClient(mode="direct", openai_client=mock_openai)
    request = {
        "model": "gpt-5-mini",
        "messages": [{"role": "user", "content": "hi"}],
        "max_completion_tokens": 256,
        "reasoning_effort": "low",
    }
    client.call(build_request_dict=request)
    call_kwargs = mock_openai.chat.completions.create.call_args[1]
    assert call_kwargs["model"] == "gpt-5-mini"
    assert call_kwargs["max_completion_tokens"] == 256
    assert call_kwargs["reasoning_effort"] == "low"

# main/server/worldmm/llm/__init__.py
# (empty)

# main/server/worldmm/llm/client.py
from __future__ import annotations

from typing import Any


def build_request(
    model: str,
    messages: list[dict[str, Any]],
    max_completion_tokens: int,
    reasoning_effort: str | None = None,
) -> dict[str, Any]:
    request: dict[str, Any] = {
        "model": model,
        "messages": messages,
        "max_completion_tokens": max_completion_tokens,
    }
    if reasoning_effort is not None:
        request["reasoning_effort"] = reasoning_effort
    return request


class LLMClient:
    def __init__(self, mode: str, openai_client: Any) -> None:
        if mode not in ("direct", "batch"):
            raise ValueError(f"Invalid mode: {mode}")
        self._mode = mode
        self._openai = openai_client

    def call(self, build_request_dict: dict[str, Any]) -> str:
        if self._mode == "direct":
            response = self._openai.chat.completions.create(**build_request_dict)
            return response.choices[0].message.content
        raise NotImplementedError("Batch mode dispatch handled by BatchClient")

git add main/server/worldmm/llm/
git commit -m "feat(worldmm): add LLM client with batch/direct dispatch"

// main/server/worldmm/tests/fixtures/sample_ner_response.json
{
  "entities": ["I", "Katrina", "dining table", "tomorrow's schedule"]
}

// main/server/worldmm/tests/fixtures/sample_triples_response.json
{
  "triples": [
    {"subject": "I", "predicate": "stand at", "object": "dining table"},
    {"subject": "Katrina", "predicate": "asks about", "object": "tomorrow's schedule"}
  ]
}

# main/server/worldmm/tests/test_openie.py
from __future__ import annotations

import json
import sys
from pathlib import Path

_server_root = Path(__file__).resolve().parents[3]
_layer_root = _server_root / "layers" / "shared" / "python"
for p in (_layer_root, _server_root):
    if str(p) not in sys.path:
        sys.path.insert(0, str(p))

FIXTURES = Path(__file__).parent / "fixtures"


def test_build_ner_prompt_includes_caption():
    from worldmm.memory.episodic.openie import build_ner_prompt
    messages = build_ner_prompt("I stand at the dining table while Katrina asks about tomorrow.")
    # Must be a list of message dicts
    assert isinstance(messages, list)
    assert any("dining table" in str(m.get("content", "")) for m in messages)


def test_parse_ner_response():
    from worldmm.memory.episodic.openie import parse_ner_response
    raw = (FIXTURES / "sample_ner_response.json").read_text()
    entities = parse_ner_response(raw)
    assert entities == ["I", "Katrina", "dining table", "tomorrow's schedule"]


def test_parse_ner_response_handles_plain_json():
    from worldmm.memory.episodic.openie import parse_ner_response
    entities = parse_ner_response('{"entities": ["Alice", "Bob"]}')
    assert entities == ["Alice", "Bob"]


def test_build_triple_prompt_includes_caption_and_entities():
    from worldmm.memory.episodic.openie import build_triple_prompt
    messages = build_triple_prompt(
        caption="I stand at the dining table",
        entities=["I", "dining table"],
    )
    assert isinstance(messages, list)
    content = str(messages)
    assert "dining table" in content
    assert "I" in content


def test_parse_triples_response():
    from worldmm.memory.episodic.openie import parse_triples_response
    raw = (FIXTURES / "sample_triples_response.json").read_text()
    triples = parse_triples_response(raw)
    assert len(triples) == 2
    assert triples[0] == {"subject": "I", "predicate": "stand at", "object": "dining table"}
    assert triples[1] == {"subject": "Katrina", "predicate": "asks about", "object": "tomorrow's schedule"}


def test_parse_triples_response_filters_invalid():
    from worldmm.memory.episodic.openie import parse_triples_response
    raw = json.dumps({"triples": [
        {"subject": "I", "predicate": "stand at", "object": "table"},
        {"subject": "I", "predicate": "does"},  # missing object — invalid
    ]})
    triples = parse_triples_response(raw)
    assert len(triples) == 1

# main/server/worldmm/memory/__init__.py
# (empty)

# main/server/worldmm/memory/episodic/__init__.py
# (empty)

# main/server/worldmm/memory/episodic/openie.py
from __future__ import annotations

import json
from pathlib import Path
from typing import Any

_PROMPTS_DIR = Path(__file__).resolve().parents[2] / "llm" / "prompts"


def _load_prompt(name: str) -> str:
    path = _PROMPTS_DIR / name
    if path.exists():
        return path.read_text().strip()
    # Fallback inline prompts for early development
    if name == "ner.txt":
        return "Extract named entities from the given paragraph. Return JSON: {\"entities\": [...]}"
    if name == "episodic_triples.txt":
        return (
            "Given the caption and extracted entities, extract RDF-style triples. "
            "Each triple must contain at least one named entity. "
            "Return JSON: {\"triples\": [{\"subject\": ..., \"predicate\": ..., \"object\": ...}, ...]}"
        )
    return ""


def build_ner_prompt(caption: str) -> list[dict[str, str]]:
    system = _load_prompt("ner.txt")
    return [
        {"role": "system", "content": system},
        {"role": "user", "content": caption},
    ]


def parse_ner_response(raw: str) -> list[str]:
    data = json.loads(raw)
    return data.get("entities", [])


def build_triple_prompt(caption: str, entities: list[str]) -> list[dict[str, str]]:
    system = _load_prompt("episodic_triples.txt")
    user_content = f"Caption: {caption}\nEntities: {json.dumps(entities)}"
    return [
        {"role": "system", "content": system},
        {"role": "user", "content": user_content},
    ]


def parse_triples_response(raw: str) -> list[dict[str, str]]:
    data = json.loads(raw)
    triples = data.get("triples", [])
    return [
        t for t in triples
        if all(k in t for k in ("subject", "predicate", "object"))
    ]

git add main/server/worldmm/memory/ main/server/worldmm/tests/test_openie.py \
        main/server/worldmm/tests/fixtures/
git commit -m "feat(worldmm): add NER + triple extraction prompt building and parsing"

# main/server/worldmm/tests/test_captioner.py
from __future__ import annotations

import base64
import sys
from pathlib import Path
from unittest.mock import MagicMock

_server_root = Path(__file__).resolve().parents[3]
_layer_root = _server_root / "layers" / "shared" / "python"
for p in (_layer_root, _server_root):
    if str(p) not in sys.path:
        sys.path.insert(0, str(p))


def test_build_caption_prompt_includes_frames_and_transcript():
    from worldmm.ingest.captioner import build_caption_prompt
    # Create 8 fake base64 frames
    frames_b64 = [base64.b64encode(f"frame{i}".encode()).decode() for i in range(8)]
    transcript = "I'm standing at the dining table"
    messages = build_caption_prompt(frames_b64, transcript)

    assert isinstance(messages, list)
    # System message + user message
    assert len(messages) == 2
    user_msg = messages[1]
    assert user_msg["role"] == "user"
    # Content should be a list (multimodal: image_url parts + text)
    assert isinstance(user_msg["content"], list)
    # 8 image parts + 1 text part
    assert len(user_msg["content"]) == 9
    # Check image parts are image_url type
    image_parts = [p for p in user_msg["content"] if p.get("type") == "image_url"]
    assert len(image_parts) == 8
    # Check text part contains transcript
    text_parts = [p for p in user_msg["content"] if p.get("type") == "text"]
    assert len(text_parts) == 1
    assert "dining table" in text_parts[0]["text"]


def test_build_caption_prompt_rejects_wrong_frame_count():
    from worldmm.ingest.captioner import build_caption_prompt
    import pytest
    with pytest.raises(ValueError, match="exactly 8"):
        build_caption_prompt(["frame1", "frame2"], "transcript")


def test_parse_caption_response():
    from worldmm.ingest.captioner import parse_caption_response
    result = parse_caption_response("The person stands at a dining table talking to Katrina.")
    assert result == "The person stands at a dining table talking to Katrina."


def test_parse_caption_response_strips_whitespace():
    from worldmm.ingest.captioner import parse_caption_response
    result = parse_caption_response("  caption text  \n")
    assert result == "caption text"

# main/server/worldmm/ingest/__init__.py
# (empty)

# main/server/worldmm/ingest/captioner.py
from __future__ import annotations

from pathlib import Path
from typing import Any

_PROMPTS_DIR = Path(__file__).resolve().parents[1] / "llm" / "prompts"


def _load_prompt(name: str) -> str:
    path = _PROMPTS_DIR / name
    if path.exists():
        return path.read_text().strip()
    if name == "caption.txt":
        return (
            "You are a video captioning assistant. Given 8 frames from a 30-second video clip "
            "and an audio transcript, generate a detailed caption describing what is happening. "
            "Include people, objects, actions, and locations."
        )
    return ""


def build_caption_prompt(
    frames_b64: list[str],
    transcript: str,
) -> list[dict[str, Any]]:
    if len(frames_b64) != 8:
        raise ValueError("Captioner requires exactly 8 frames")

    system = _load_prompt("caption.txt")
    content_parts: list[dict[str, Any]] = []
    for frame in frames_b64:
        content_parts.append({
            "type": "image_url",
            "image_url": {"url": f"data:image/jpeg;base64,{frame}"},
        })
    content_parts.append({
        "type": "text",
        "text": f"Audio transcript: {transcript}",
    })

    return [
        {"role": "system", "content": system},
        {"role": "user", "content": content_parts},
    ]


def parse_caption_response(raw: str) -> str:
    return raw.strip()

git add main/server/worldmm/ingest/ main/server/worldmm/tests/test_captioner.py
git commit -m "feat(worldmm): add captioner prompt building with multimodal frames"

# main/server/worldmm/tests/test_multiscale.py
from __future__ import annotations

import sys
from pathlib import Path

_server_root = Path(__file__).resolve().parents[3]
_layer_root = _server_root / "layers" / "shared" / "python"
for p in (_layer_root, _server_root):
    if str(p) not in sys.path:
        sys.path.insert(0, str(p))


def test_build_merge_prompt_includes_all_captions():
    from worldmm.memory.episodic.multiscale import build_merge_prompt
    captions = [
        "Caption 1: I enter the kitchen",
        "Caption 2: I open the fridge",
        "Caption 3: I take out milk",
        "Caption 4: I pour milk into a glass",
        "Caption 5: I drink the milk",
        "Caption 6: I wash the glass",
    ]
    messages = build_merge_prompt(captions, target_duration_seconds=180)
    assert isinstance(messages, list)
    content = str(messages)
    for cap in captions:
        assert cap in content
    assert "3-minute" in content or "180" in content


def test_build_merge_prompt_for_hour_scale():
    from worldmm.memory.episodic.multiscale import build_merge_prompt
    summaries = [f"Summary {i}" for i in range(6)]
    messages = build_merge_prompt(summaries, target_duration_seconds=3600)
    content = str(messages)
    assert "1-hour" in content or "3600" in content


def test_parse_merge_response():
    from worldmm.memory.episodic.multiscale import parse_merge_response
    result = parse_merge_response("The person went to the kitchen and made milk.")
    assert result == "The person went to the kitchen and made milk."


def test_build_merge_prompt_rejects_empty():
    from worldmm.memory.episodic.multiscale import build_merge_prompt
    import pytest
    with pytest.raises(ValueError, match="at least one"):
        build_merge_prompt([], target_duration_seconds=180)

# main/server/worldmm/memory/episodic/multiscale.py
from __future__ import annotations

from pathlib import Path
from typing import Any

_PROMPTS_DIR = Path(__file__).resolve().parents[2] / "llm" / "prompts"

_DURATION_LABELS = {
    180: "3-minute",
    600: "10-minute",
    3600: "1-hour",
}


def _load_prompt(name: str) -> str:
    path = _PROMPTS_DIR / name
    if path.exists():
        return path.read_text().strip()
    if name == "merge_captions.txt":
        return (
            "You are a summarization assistant. Given a sequence of captions from consecutive "
            "video segments, merge them into a single coherent {duration_label} summary. "
            "Preserve key entities, actions, and temporal order. Be concise but complete."
        )
    return ""


def build_merge_prompt(
    captions: list[str],
    target_duration_seconds: int,
) -> list[dict[str, str]]:
    if not captions:
        raise ValueError("Merge requires at least one caption")

    label = _DURATION_LABELS.get(target_duration_seconds, f"{target_duration_seconds}-second")
    template = _load_prompt("merge_captions.txt")
    system = template.replace("{duration_label}", label)

    numbered = "\n".join(f"{i + 1}. {c}" for i, c in enumerate(captions))
    user_content = f"Merge these captions into a {label} summary:\n\n{numbered}"

    return [
        {"role": "system", "content": system},
        {"role": "user", "content": user_content},
    ]


def parse_merge_response(raw: str) -> str:
    return raw.strip()

git add main/server/worldmm/memory/episodic/multiscale.py \
        main/server/worldmm/tests/test_multiscale.py
git commit -m "feat(worldmm): add multi-scale caption merging prompts"

# main/server/worldmm/tests/test_episodic_graph.py
from __future__ import annotations

import sys
from pathlib import Path

_server_root = Path(__file__).resolve().parents[3]
_layer_root = _server_root / "layers" / "shared" / "python"
for p in (_layer_root, _server_root):
    if str(p) not in sys.path:
        sys.path.insert(0, str(p))


def _make_triples():
    """Create a small knowledge graph for testing PPR."""
    return [
        {"subject_entity_id": "e1", "predicate": "stand at", "object_entity_id": "e2",
         "object_literal": None, "segment_id": "s1", "id": "t1"},
        {"subject_entity_id": "e1", "predicate": "talks to", "object_entity_id": "e3",
         "object_literal": None, "segment_id": "s1", "id": "t2"},
        {"subject_entity_id": "e3", "predicate": "asks about", "object_entity_id": None,
         "object_literal": "tomorrow's schedule", "segment_id": "s1", "id": "t3"},
    ]


def _make_entity_names():
    return {"e1": "I", "e2": "dining table", "e3": "Katrina"}


def _make_segment_captions():
    return {"s1": "I stand at the dining table while Katrina asks about tomorrow's schedule"}


def test_build_graph_creates_nodes():
    from worldmm.memory.episodic.graph import EpisodicGraph
    g = EpisodicGraph()
    g.build(_make_triples(), _make_entity_names())
    # 3 entity nodes + 1 literal node
    assert g.node_count() == 4


def test_build_graph_creates_edges():
    from worldmm.memory.episodic.graph import EpisodicGraph
    g = EpisodicGraph()
    g.build(_make_triples(), _make_entity_names())
    assert g.edge_count() == 3


def test_ppr_from_seed_entity():
    from worldmm.memory.episodic.graph import EpisodicGraph
    g = EpisodicGraph()
    g.build(_make_triples(), _make_entity_names())
    ranked = g.ppr(seed_entity_ids=["e1"], top_k=5)
    # e1 is the seed, so it should have highest PPR score
    assert ranked[0]["entity_id"] == "e1"
    # All connected entities should appear
    entity_ids = {r["entity_id"] for r in ranked}
    assert "e3" in entity_ids  # directly connected


def test_ppr_returns_segment_ids():
    from worldmm.memory.episodic.graph import EpisodicGraph
    g = EpisodicGraph()
    g.build(_make_triples(), _make_entity_names())
    ranked = g.ppr(seed_entity_ids=["e1"], top_k=5)
    # Each result should include the segment_ids that reference it
    for r in ranked:
        assert "segment_ids" in r


def test_get_captions_for_segments():
    from worldmm.memory.episodic.graph import EpisodicGraph
    g = EpisodicGraph()
    g.build(_make_triples(), _make_entity_names())
    g.set_segment_captions(_make_segment_captions())
    captions = g.get_captions(["s1"])
    assert captions == ["I stand at the dining table while Katrina asks about tomorrow's schedule"]


def test_empty_graph_ppr_returns_empty():
    from worldmm.memory.episodic.graph import EpisodicGraph
    g = EpisodicGraph()
    g.build([], {})
    ranked = g.ppr(seed_entity_ids=["nonexistent"], top_k=5)
    assert ranked == []

# main/server/worldmm/memory/episodic/graph.py
from __future__ import annotations

from collections import defaultdict
from typing import Any

import igraph as ig


class EpisodicGraph:
    def __init__(self) -> None:
        self._graph: ig.Graph | None = None
        self._node_id_to_idx: dict[str, int] = {}
        self._idx_to_node_id: dict[int, str] = {}
        self._node_segment_ids: dict[str, set[str]] = defaultdict(set)
        self._segment_captions: dict[str, str] = {}

    def build(
        self,
        triples: list[dict[str, Any]],
        entity_names: dict[str, str],
    ) -> None:
        nodes: set[str] = set()
        edges: list[tuple[str, str]] = []

        for t in triples:
            subj = t["subject_entity_id"]
            nodes.add(subj)
            seg_id = t.get("segment_id")

            if t.get("object_entity_id"):
                obj = t["object_entity_id"]
                nodes.add(obj)
                edges.append((subj, obj))
                if seg_id:
                    self._node_segment_ids[subj].add(seg_id)
                    self._node_segment_ids[obj].add(seg_id)
            elif t.get("object_literal"):
                literal_id = f"lit:{t['id']}"
                nodes.add(literal_id)
                edges.append((subj, literal_id))
                if seg_id:
                    self._node_segment_ids[subj].add(seg_id)
                    self._node_segment_ids[literal_id].add(seg_id)

        if not nodes:
            self._graph = ig.Graph(directed=False)
            return

        node_list = sorted(nodes)
        self._node_id_to_idx = {nid: i for i, nid in enumerate(node_list)}
        self._idx_to_node_id = {i: nid for nid, i in self._node_id_to_idx.items()}

        g = ig.Graph(n=len(node_list), directed=False)
        g.vs["node_id"] = node_list
        g.vs["name"] = [entity_names.get(nid, nid) for nid in node_list]

        edge_indices = []
        for src, dst in edges:
            if src in self._node_id_to_idx and dst in self._node_id_to_idx:
                edge_indices.append((self._node_id_to_idx[src], self._node_id_to_idx[dst]))
        g.add_edges(edge_indices)
        self._graph = g

    def node_count(self) -> int:
        return self._graph.vcount() if self._graph else 0

    def edge_count(self) -> int:
        return self._graph.ecount() if self._graph else 0

    def ppr(
        self,
        seed_entity_ids: list[str],
        top_k: int = 10,
        damping: float = 0.85,
    ) -> list[dict[str, Any]]:
        if not self._graph or self._graph.vcount() == 0:
            return []

        reset_vec = [0.0] * self._graph.vcount()
        valid_seeds = [eid for eid in seed_entity_ids if eid in self._node_id_to_idx]
        if not valid_seeds:
            return []

        weight = 1.0 / len(valid_seeds)
        for eid in valid_seeds:
            reset_vec[self._node_id_to_idx[eid]] = weight

        scores = self._graph.personalized_pagerank(
            reset=reset_vec,
            damping=damping,
            implementation="prpack",
        )

        ranked = []
        for idx, score in enumerate(scores):
            node_id = self._idx_to_node_id[idx]
            ranked.append({
                "entity_id": node_id,
                "name": self._graph.vs[idx]["name"],
                "score": score,
                "segment_ids": sorted(self._node_segment_ids.get(node_id, set())),
            })
        ranked.sort(key=lambda x: x["score"], reverse=True)
        return ranked[:top_k]

    def set_segment_captions(self, captions: dict[str, str]) -> None:
        self._segment_captions = captions

    def get_captions(self, segment_ids: list[str]) -> list[str]:
        return [self._segment_captions[sid] for sid in segment_ids if sid in self._segment_captions]

git add main/server/worldmm/memory/episodic/graph.py \
        main/server/worldmm/tests/test_episodic_graph.py
git commit -m "feat(worldmm): add episodic graph with PPR via igraph"

# main/server/worldmm/tests/test_semantic_graph.py
from __future__ import annotations

import sys
from pathlib import Path

_server_root = Path(__file__).resolve().parents[3]
_layer_root = _server_root / "layers" / "shared" / "python"
for p in (_layer_root, _server_root):
    if str(p) not in sys.path:
        sys.path.insert(0, str(p))


def _make_semantic_triples():
    return [
        {"id": "t1", "subject_entity_id": "e1", "predicate": "often eats",
         "object_entity_id": "e2", "object_literal": None},
        {"id": "t2", "subject_entity_id": "e1", "predicate": "is friends with",
         "object_entity_id": "e3", "object_literal": None},
        {"id": "t3", "subject_entity_id": "e3", "predicate": "helps with",
         "object_entity_id": None, "object_literal": "expense tracking"},
    ]


def _make_entity_names():
    return {"e1": "I", "e2": "fruits", "e3": "Alice"}


def test_build_semantic_graph():
    from worldmm.memory.semantic.graph import SemanticGraph
    g = SemanticGraph()
    g.build(_make_semantic_triples(), _make_entity_names())
    assert g.node_count() == 4  # 3 entities + 1 literal


def test_ppr_edge_scoring():
    from worldmm.memory.semantic.graph import SemanticGraph
    g = SemanticGraph()
    g.build(_make_semantic_triples(), _make_entity_names())
    ranked_triples = g.ppr_edge_scored(seed_entity_ids=["e1"], top_k=10)
    assert len(ranked_triples) > 0
    # Each result is a triple with a score
    for t in ranked_triples:
        assert "triple_id" in t
        assert "score" in t
        assert "subject" in t
        assert "predicate" in t


def test_ppr_edge_score_is_sum_of_node_scores():
    from worldmm.memory.semantic.graph import SemanticGraph
    g = SemanticGraph()
    triples = [
        {"id": "t1", "subject_entity_id": "e1", "predicate": "knows",
         "object_entity_id": "e2", "object_literal": None},
    ]
    g.build(triples, {"e1": "A", "e2": "B"})
    ranked = g.ppr_edge_scored(seed_entity_ids=["e1"], top_k=10)
    # Score should be PPR(subject) + PPR(object)
    node_scores = g.ppr_node_scores(seed_entity_ids=["e1"])
    expected = node_scores.get("e1", 0) + node_scores.get("e2", 0)
    assert abs(ranked[0]["score"] - expected) < 1e-6


def test_empty_semantic_graph():
    from worldmm.memory.semantic.graph import SemanticGraph
    g = SemanticGraph()
    g.build([], {})
    ranked = g.ppr_edge_scored(seed_entity_ids=["e1"], top_k=10)
    assert ranked == []

# main/server/worldmm/memory/semantic/__init__.py
# (empty)

# main/server/worldmm/memory/semantic/graph.py
from __future__ import annotations

from collections import defaultdict
from typing import Any

import igraph as ig


class SemanticGraph:
    def __init__(self) -> None:
        self._graph: ig.Graph | None = None
        self._node_id_to_idx: dict[str, int] = {}
        self._idx_to_node_id: dict[int, str] = {}
        self._triples: list[dict[str, Any]] = []
        self._entity_names: dict[str, str] = {}

    def build(
        self,
        triples: list[dict[str, Any]],
        entity_names: dict[str, str],
    ) -> None:
        self._triples = triples
        self._entity_names = entity_names
        nodes: set[str] = set()
        edges: list[tuple[str, str]] = []

        for t in triples:
            subj = t["subject_entity_id"]
            nodes.add(subj)
            if t.get("object_entity_id"):
                obj = t["object_entity_id"]
                nodes.add(obj)
                edges.append((subj, obj))
            elif t.get("object_literal"):
                literal_id = f"lit:{t['id']}"
                nodes.add(literal_id)
                edges.append((subj, literal_id))

        if not nodes:
            self._graph = ig.Graph(directed=False)
            return

        node_list = sorted(nodes)
        self._node_id_to_idx = {nid: i for i, nid in enumerate(node_list)}
        self._idx_to_node_id = {i: nid for nid, i in self._node_id_to_idx.items()}

        g = ig.Graph(n=len(node_list), directed=False)
        g.vs["node_id"] = node_list
        g.vs["name"] = [entity_names.get(nid, nid) for nid in node_list]

        edge_indices = []
        for src, dst in edges:
            if src in self._node_id_to_idx and dst in self._node_id_to_idx:
                edge_indices.append((self._node_id_to_idx[src], self._node_id_to_idx[dst]))
        g.add_edges(edge_indices)
        self._graph = g

    def node_count(self) -> int:
        return self._graph.vcount() if self._graph else 0

    def ppr_node_scores(
        self,
        seed_entity_ids: list[str],
        damping: float = 0.85,
    ) -> dict[str, float]:
        if not self._graph or self._graph.vcount() == 0:
            return {}

        reset_vec = [0.0] * self._graph.vcount()
        valid_seeds = [eid for eid in seed_entity_ids if eid in self._node_id_to_idx]
        if not valid_seeds:
            return {}

        weight = 1.0 / len(valid_seeds)
        for eid in valid_seeds:
            reset_vec[self._node_id_to_idx[eid]] = weight

        scores = self._graph.personalized_pagerank(
            reset=reset_vec,
            damping=damping,
            implementation="prpack",
        )
        return {self._idx_to_node_id[i]: s for i, s in enumerate(scores)}

    def ppr_edge_scored(
        self,
        seed_entity_ids: list[str],
        top_k: int = 10,
        damping: float = 0.85,
    ) -> list[dict[str, Any]]:
        node_scores = self.ppr_node_scores(seed_entity_ids, damping)
        if not node_scores:
            return []

        results = []
        for t in self._triples:
            subj = t["subject_entity_id"]
            if t.get("object_entity_id"):
                obj = t["object_entity_id"]
            elif t.get("object_literal"):
                obj = f"lit:{t['id']}"
            else:
                continue

            score = node_scores.get(subj, 0.0) + node_scores.get(obj, 0.0)
            obj_display = (
                self._entity_names.get(t["object_entity_id"], t.get("object_entity_id", ""))
                if t.get("object_entity_id")
                else t.get("object_literal", "")
            )
            results.append({
                "triple_id": t["id"],
                "score": score,
                "subject": self._entity_names.get(subj, subj),
                "predicate": t["predicate"],
                "object": obj_display,
            })

        results.sort(key=lambda x: x["score"], reverse=True)
        return results[:top_k]