wip(hindsight-dev): mission-sandbox tool for generating better observation missions

Restored from stash 'wip-mission-sandbox'. WIP backup.

hindsight-api-slim/hindsight_api/engine/memory_engine.py

@@ -16,7 +16,7 @@ import logging
 import time
 import uuid
 from collections.abc import Awaitable, Callable
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from datetime import UTC, datetime, timedelta, timezone
 from typing import TYPE_CHECKING, Any, Literal, cast, overload
 
@@ -273,10 +273,18 @@ class _SubBatchSplit:
     user (such as ``retain_batch_async``) use this mapping to merge
     results belonging to the same original content back together when
     an oversized item was chunked across multiple sub-batches.
+
+    ``document_body_overrides[i]`` is the full original body of the
+    oversized item that produced ``sub_batches[i]``, or ``None`` when
+    the sub-batch was not produced by chunking an oversized item. The
+    orchestrator uses this as the ``documents.original_text`` payload
+    so that slicing an item across sub-batches does not persist a
+    partial body (see issue #1838).
     """
 
     sub_batches: list[list[RetainContentDict]]
     origin_indices: list[list[int]]
+    document_body_overrides: list[str | None] = field(default_factory=list)
 
 
 def _split_contents_into_sub_batches(
@@ -310,6 +318,7 @@ def _split_contents_into_sub_batches(
 
     sub_batches: list[list[RetainContentDict]] = []
     origin_indices: list[list[int]] = []
+    document_body_overrides: list[str | None] = []
     current_batch: list[RetainContentDict] = []
     current_batch_origins: list[int] = []
     current_batch_tokens = 0
@@ -319,6 +328,7 @@ def _split_contents_into_sub_batches(
         if current_batch:
             sub_batches.append(current_batch)
             origin_indices.append(current_batch_origins)
+            document_body_overrides.append(None)
             current_batch = []
             current_batch_origins = []
             current_batch_tokens = 0
@@ -334,12 +344,18 @@ def _split_contents_into_sub_batches(
             # original item's document_id and metadata so the
             # orchestrator's first-batch document tracking still
             # cascade-deletes the prior document version on slice 1.
+            # Each slice carries ``content_str`` as the document body
+            # override so the orchestrator writes the full original
+            # text to documents.original_text — not just its own slice
+            # (otherwise the last slice would clobber the body with a
+            # truncated payload; see issue #1838).
             _flush()
             chunks = fact_extraction.chunk_text(content_str, char_budget)
             for chunk in chunks:
                 chunk_item = cast(RetainContentDict, {**item, "content": chunk})
                 sub_batches.append([chunk_item])
                 origin_indices.append([original_idx])
+                document_body_overrides.append(content_str)
             continue
 
         if current_batch and current_batch_tokens + item_tokens > tokens_per_batch:
@@ -349,7 +365,11 @@ def _split_contents_into_sub_batches(
         current_batch_tokens += item_tokens
 
     _flush()
-    return _SubBatchSplit(sub_batches=sub_batches, origin_indices=origin_indices)
+    return _SubBatchSplit(
+        sub_batches=sub_batches,
+        origin_indices=origin_indices,
+        document_body_overrides=document_body_overrides,
+    )
 
 
 def _split_contents_into_async_children(
@@ -2740,6 +2760,7 @@ class MemoryEngine(MemoryEngineInterface):
             split = _split_contents_into_sub_batches(contents, tokens_per_batch)
             sub_batches = split.sub_batches
             origin_indices = split.origin_indices
+            document_body_overrides = split.document_body_overrides
 
             sub_batch_sizes = [len(b) for b in sub_batches]
             # Keep the per-sub-batch sizes log compact when an oversize
@@ -2787,6 +2808,7 @@ class MemoryEngine(MemoryEngineInterface):
                     # webhook delivery row is committed atomically with the final retain data.
                     outbox_callback=outbox_callback if i == len(sub_batches) else None,
                     outbox_callback_factory=outbox_callback_factory if i == len(sub_batches) else None,
+                    document_body_override=document_body_overrides[i - 1],
                 )
                 # sub_results aligns 1:1 with sub_batch items; map each
                 # back to its source input via origin_indices so callers
@@ -2880,6 +2902,7 @@ class MemoryEngine(MemoryEngineInterface):
         outbox_callback: RetainOutboxCallback | None = None,
         outbox_callback_factory: RetainOutboxCallbackFactory | None = None,
         strategy: str | None = None,
+        document_body_override: str | None = None,
     ) -> tuple[list[list[str]], "TokenUsage", int | None]:
         """
         Internal method for batch processing without chunking logic.
@@ -2943,6 +2966,7 @@ class MemoryEngine(MemoryEngineInterface):
                 outbox_callback=outbox_callback,
                 outbox_callback_factory=outbox_callback_factory,
                 db_semaphore=self._put_semaphore,
+                document_body_override=document_body_override,
             )
 
     def recall(

hindsight-api-slim/hindsight_api/engine/retain/orchestrator.py

@@ -402,6 +402,7 @@ async def retain_batch(
     outbox_callback: RetainOutboxCallback | None = None,
     outbox_callback_factory: RetainOutboxCallbackFactory | None = None,
     db_semaphore: "asyncio.Semaphore | None" = None,
+    document_body_override: str | None = None,
 ) -> tuple[list[list[str]], TokenUsage, int | None]:
     """
     Process a batch of content through the retain pipeline.
@@ -480,6 +481,7 @@ async def retain_batch(
                     outbox_callback=group_outbox_callback,
                     outbox_callback_factory=outbox_callback_factory,
                     db_semaphore=db_semaphore,
+                    document_body_override=document_body_override,
                 )
                 for group_idx, orig_idx in enumerate(original_indices[doc_key]):
                     if group_idx < len(group_ids):
@@ -621,6 +623,7 @@ async def retain_batch(
             schema,
             outbox_callback,
             db_semaphore,
+            document_body_override=document_body_override,
         )
         if delta_result is not None:
             return delta_result
@@ -677,6 +680,7 @@ async def retain_batch(
         schema=schema,
         outbox_callback=outbox_callback,
         db_semaphore=db_semaphore,
+        document_body_override=document_body_override,
     )
 
 
@@ -814,6 +818,7 @@ async def _streaming_retain_batch(
     schema: str | None = None,
     outbox_callback: Callable[["asyncpg.Connection"], Awaitable[None]] | None = None,
     db_semaphore: "asyncio.Semaphore | None" = None,
+    document_body_override: str | None = None,
 ) -> tuple[list[list[str]], TokenUsage]:
     """
     Process a large document in streaming mini-batches to bound memory usage.
@@ -847,7 +852,15 @@ async def _streaming_retain_batch(
     # document exists with a matching content_hash and has committed chunks,
     # the producer can skip already-extracted chunks to avoid duplicate work.
     existing_chunk_hashes: set[str] = set()
-    combined_content = "\n".join([c.get("content", "") for c in contents_dicts])
+    # When the caller is processing a sub-batch sliced out of an oversized
+    # item (see _split_contents_into_sub_batches), document_body_override
+    # carries the full original document body. Use it for the doc-row write
+    # so documents.original_text stores the complete payload, not just this
+    # slice (issue #1838).
+    if document_body_override is not None:
+        combined_content = document_body_override
+    else:
+        combined_content = "\n".join([c.get("content", "") for c in contents_dicts])
     # Memory: contents_dicts content strings are now captured in combined_content.
     # Clear them from the dicts to release the per-item copies (can be multi-MB each).
     for d in contents_dicts:
@@ -1490,6 +1503,8 @@ async def _try_delta_retain(
     schema,
     outbox_callback,
     db_semaphore: "asyncio.Semaphore | None" = None,
+    *,
+    document_body_override: str | None = None,
 ) -> tuple[list[list[str]], TokenUsage, int | None] | None:
     """
     Attempt delta retain for a document upsert. Returns result tuple if delta
@@ -1575,6 +1590,7 @@ async def _try_delta_retain(
             log_buffer,
             start_time,
             outbox_callback,
+            document_body_override=document_body_override,
         )
 
     # Build content items for only the changed/new chunks
@@ -1591,6 +1607,7 @@ async def _try_delta_retain(
             log_buffer,
             start_time,
             outbox_callback,
+            document_body_override=document_body_override,
         )
 
     # Extract facts and generate embeddings (shared pipeline)
@@ -1650,7 +1667,13 @@ async def _try_delta_retain(
 
                 # Update document metadata (no delete)
                 step_start = time.time()
-                combined_content = "\n".join([c.get("content", "") for c in contents_dicts])
+                # When this sub-batch is one slice of an oversized item
+                # split across multiple sub-batches, store the full body
+                # (issue #1838) instead of just the slice.
+                if document_body_override is not None:
+                    combined_content = document_body_override
+                else:
+                    combined_content = "\n".join([c.get("content", "") for c in contents_dicts])
                 retain_params, merged_tags = _build_retain_params(contents_dicts, document_tags)
                 await fact_storage.upsert_document_metadata(
                     conn,
@@ -1775,6 +1798,8 @@ async def _delta_metadata_only(
     log_buffer,
     start_time,
     outbox_callback,
+    *,
+    document_body_override: str | None = None,
 ):
     """Handle the case where no chunks changed — just update document metadata and tags."""
     async with acquire_with_retry(pool) as conn:
@@ -1785,7 +1810,12 @@ async def _delta_metadata_only(
                 document_id,
                 bank_id,
             )
-            combined_content = "\n".join([c.get("content", "") for c in contents_dicts])
+            # When this sub-batch is a slice of an oversized item, write the
+            # full original body (issue #1838) instead of just the slice.
+            if document_body_override is not None:
+                combined_content = document_body_override
+            else:
+                combined_content = "\n".join([c.get("content", "") for c in contents_dicts])
             retain_params, merged_tags = _build_retain_params(contents_dicts, document_tags)
             await fact_storage.upsert_document_metadata(
                 conn,

hindsight-api-slim/tests/test_large_document_replacement.py

@@ -0,0 +1,157 @@
+"""
+Reproduction for https://github.com/vectorize-io/hindsight/issues/1838
+
+Replacing an existing document by retaining new content with the same
+``document_id`` can leave the stored document body partial/truncated when
+``retain_batch_async`` auto-splits the submitted content into multiple
+sub-batches. Each non-first sub-batch was overwriting
+``documents.original_text`` with its own slice, so the persisted body ended
+up being one slice of the input, not the full body.
+
+These tests trigger the auto-split path by lowering
+``HINDSIGHT_API_RETAIN_BATCH_TOKENS`` to a small value, then assert that the
+stored ``original_text`` exactly matches the submitted replacement body.
+"""
+
+from datetime import datetime, timezone
+
+import pytest
+
+from hindsight_api.config import clear_config_cache
+
+
+def _ts() -> float:
+    return datetime.now(timezone.utc).timestamp()
+
+
+@pytest.fixture(autouse=True)
+def _fast_split_env(monkeypatch):
+    """Make the splitter trigger on small content and skip consolidation work.
+
+    Auto-consolidation runs synchronously after each retain in tests; it
+    extracts/recalls/embeds across all observations and dominates wall time
+    for these tests, which only care about how the splitter persists the
+    document body. Disabling it brings the suite back to single-digit
+    seconds.
+    """
+    monkeypatch.setenv("HINDSIGHT_API_RETAIN_BATCH_TOKENS", "100")
+    monkeypatch.setenv("HINDSIGHT_API_ENABLE_AUTO_CONSOLIDATION", "false")
+    monkeypatch.setenv("HINDSIGHT_API_ENABLE_OBSERVATIONS", "false")
+    clear_config_cache()
+    yield
+    clear_config_cache()
+
+
+def _make_replacement_body() -> str:
+    """Build a multi-line body comfortably above the 100-token splitter
+    threshold so ``_split_contents_into_sub_batches`` chunks it into more
+    than one sub-batch.
+    """
+    lines = [
+        f"[role: user] turn {i}: alpha bravo charlie delta echo "
+        f"foxtrot golf hotel india juliet"
+        for i in range(20)
+    ]
+    return "\n".join(lines)
+
+
+@pytest.mark.asyncio
+async def test_large_same_id_replacement_preserves_full_body(memory, request_context):
+    """
+    RED test for issue #1838.
+
+    Retain a small initial document, then replace it with a larger body
+    under the same ``document_id``. The replacement is sized to trip the
+    ``retain_batch_tokens`` threshold so ``retain_batch_async`` splits it
+    into multiple sub-batches. After retain returns, the stored
+    ``original_text`` must exactly equal the submitted replacement body.
+    """
+    bank_id = f"test_large_replace_{_ts()}"
+    document_id = "claude-code-transcript-1838"
+
+    try:
+        initial_body = "[role: user] turn 0: hello\n[role: assistant] turn 0: hi"
+        await memory.retain_async(
+            bank_id=bank_id,
+            content=initial_body,
+            context="initial retro",
+            document_id=document_id,
+            request_context=request_context,
+        )
+
+        doc_initial = await memory.get_document(
+            document_id, bank_id, request_context=request_context
+        )
+        assert doc_initial is not None
+        assert doc_initial["original_text"] == initial_body
+
+        replacement_body = _make_replacement_body()
+
+        await memory.retain_async(
+            bank_id=bank_id,
+            content=replacement_body,
+            context="regenerated retro",
+            document_id=document_id,
+            request_context=request_context,
+        )
+
+        doc_replaced = await memory.get_document(
+            document_id, bank_id, request_context=request_context
+        )
+        assert doc_replaced is not None
+
+        stored = doc_replaced["original_text"]
+        assert len(stored) == len(replacement_body), (
+            f"stored body length {len(stored)} != submitted length "
+            f"{len(replacement_body)} — partial replacement persisted"
+        )
+        assert stored == replacement_body, (
+            "stored original_text does not exactly match the submitted "
+            "replacement body"
+        )
+
+    finally:
+        await memory.delete_bank(bank_id, request_context=request_context)
+
+
+@pytest.mark.asyncio
+async def test_repeated_large_same_id_replacement_is_idempotent(memory, request_context):
+    """
+    Retrying the same large replacement (per issue #1838 acceptance criteria)
+    must converge to the exact submitted body, not a suffix/prefix subset.
+    """
+    bank_id = f"test_large_replace_retry_{_ts()}"
+    document_id = "claude-code-transcript-1838-retry"
+
+    try:
+        await memory.retain_async(
+            bank_id=bank_id,
+            content="[role: user] turn 0: seed",
+            context="seed",
+            document_id=document_id,
+            request_context=request_context,
+        )
+
+        replacement_body = _make_replacement_body()
+
+        for attempt in range(3):
+            await memory.retain_async(
+                bank_id=bank_id,
+                content=replacement_body,
+                context=f"regenerated retro attempt {attempt}",
+                document_id=document_id,
+                request_context=request_context,
+            )
+
+            doc = await memory.get_document(
+                document_id, bank_id, request_context=request_context
+            )
+            assert doc is not None, f"attempt {attempt}: document missing after retain"
+            assert doc["original_text"] == replacement_body, (
+                f"attempt {attempt}: stored body diverged from submitted body "
+                f"(stored {len(doc['original_text'])} chars, "
+                f"submitted {len(replacement_body)} chars)"
+            )
+
+    finally:
+        await memory.delete_bank(bank_id, request_context=request_context)

hindsight-dev/hindsight_dev/mission_sandbox/agent_labeler.py

@@ -0,0 +1,92 @@
+"""agent-label command: LLM-based observation labeling."""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+from .helpers import console
+from .project import LabelStep, Project
+
+_LABEL_SYSTEM_PROMPT = """\
+You are evaluating observations generated by a memory system.
+Each observation was auto-generated from raw facts extracted from documents.
+
+The user will tell you what kind of observations they consider valuable.
+For each observation, decide whether it is "good" or "bad" and explain why in one sentence.
+
+Respond with JSON: {"label": "good" | "bad", "reason": "..."}
+"""
+
+
+def _label_one(observation_text: str, source_facts: list[str], instructions: str, model: str) -> tuple[str, str]:
+    """Call LLM to label a single observation."""
+    import litellm
+
+    facts_str = "\n".join(f"- {f}" for f in source_facts) if source_facts else "(no source facts available)"
+
+    user_msg = f"""\
+## User instructions
+{instructions}
+
+## Observation
+{observation_text}
+
+## Source facts
+{facts_str}
+"""
+
+    response = litellm.completion(
+        model=model,
+        messages=[
+            {"role": "system", "content": _LABEL_SYSTEM_PROMPT},
+            {"role": "user", "content": user_msg},
+        ],
+        response_format={"type": "json_object"},
+        temperature=0,
+    )
+    raw = response.choices[0].message.content
+    parsed = json.loads(raw)
+    return parsed["label"], parsed["reason"]
+
+
+def run_agent_label(
+    project_path: Path,
+    instructions: str,
+    model: str,
+) -> None:
+    proj = Project.load(project_path)
+
+    if not proj.observations:
+        console.print("[red]No observations in project. Run init first.[/red]")
+        return
+
+    unlabeled = [o for o in proj.observations if o.label is None]
+    if not unlabeled:
+        console.print("[yellow]All observations are already labeled.[/yellow]")
+        return
+
+    console.print(f"[bold]Labeling {len(unlabeled)} observations with [cyan]{model}[/cyan]...[/bold]")
+
+    for i, obs in enumerate(unlabeled, 1):
+        console.print(f"\n[dim]({i}/{len(unlabeled)})[/dim] {obs.text[:100]}...")
+        label, reason = _label_one(obs.text, obs.source_facts, instructions, model)
+        obs.label = label
+        obs.reason = reason
+        console.print(f"  [{('green' if label == 'good' else 'red')}]{label}[/] — {reason}")
+        # Save incrementally
+        proj.save()
+
+    good = sum(1 for o in proj.observations if o.label == "good")
+    bad = sum(1 for o in proj.observations if o.label == "bad")
+
+    # Record history
+    step = LabelStep(
+        instructions=instructions,
+        model=model,
+        observations=list(proj.observations),
+    )
+    step_path = proj.add_step(step)
+
+    console.print(f"\n[green bold]Done![/green bold] {good} good, {bad} bad out of {len(proj.observations)} total")
+    console.print(f"  History: {step_path.name}")

hindsight-dev/hindsight_dev/mission_sandbox/cli.py

@@ -0,0 +1,123 @@
+"""CLI entry point for mission-sandbox."""
+
+from __future__ import annotations
+
+import argparse
+import sys
+from pathlib import Path
+
+from rich.console import Console
+
+console = Console()
+
+
+def cmd_init(args: argparse.Namespace) -> None:
+    from .init import run_init
+
+    run_init(
+        project_path=Path(args.project),
+        documents_path=Path(args.documents),
+        bank_id=args.bank_id,
+        api_url=args.api_url,
+    )
+
+
+def cmd_agent_label(args: argparse.Namespace) -> None:
+    from .agent_labeler import run_agent_label
+
+    run_agent_label(
+        project_path=Path(args.project),
+        instructions=args.instructions,
+        model=args.model,
+    )
+
+
+def cmd_optimize(args: argparse.Namespace) -> None:
+    from .optimizer import run_optimize
+
+    run_optimize(
+        project_path=Path(args.project),
+        model=args.model,
+    )
+
+
+def cmd_run(args: argparse.Namespace) -> None:
+    from .run import run_run
+
+    run_run(project_path=Path(args.project))
+
+
+def cmd_promote(args: argparse.Namespace) -> None:
+    from .promote import run_promote
+
+    run_promote(
+        project_path=Path(args.project),
+        target_bank=args.target_bank,
+        backfill=args.backfill,
+    )
+
+
+def cmd_ui(args: argparse.Namespace) -> None:
+    import subprocess
+
+    ui_script = Path(__file__).parent / "ui.py"
+    subprocess.run(
+        ["streamlit", "run", str(ui_script), "--", str(Path(args.project).resolve())],
+        check=True,
+    )
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(
+        prog="mission-sandbox",
+        description="Iterate on Hindsight observation missions with a fast feedback loop.",
+    )
+    sub = parser.add_subparsers(dest="command", required=True)
+
+    # init
+    p_init = sub.add_parser("init", help="Create project, ingest documents, run baseline consolidation")
+    p_init.add_argument("project", help="Project directory to create")
+    p_init.add_argument("--documents", required=True, help="Path to documents dir or file")
+    p_init.add_argument("--bank-id", required=True, help="Sandbox bank ID")
+    p_init.add_argument("--api-url", default="http://localhost:8888", help="Hindsight API URL")
+    p_init.set_defaults(func=cmd_init)
+
+    # agent-label
+    p_label = sub.add_parser("agent-label", help="LLM-based observation labeling")
+    p_label.add_argument("project", help="Project directory")
+    p_label.add_argument("--instructions", required=True, help="What observations you want")
+    p_label.add_argument("--model", default="anthropic/claude-sonnet-4-20250514", help="LiteLLM model")
+    p_label.set_defaults(func=cmd_agent_label)
+
+    # optimize
+    p_opt = sub.add_parser("optimize", help="Generate improved mission from labeled observations")
+    p_opt.add_argument("project", help="Project directory")
+    p_opt.add_argument("--model", default="anthropic/claude-sonnet-4-20250514", help="LiteLLM model")
+    p_opt.set_defaults(func=cmd_optimize)
+
+    # run
+    p_run = sub.add_parser("run", help="Apply mission, re-consolidate, update observations for next label round")
+    p_run.add_argument("project", help="Project directory")
+    p_run.set_defaults(func=cmd_run)
+
+    # promote
+    p_prom = sub.add_parser("promote", help="Push optimized mission to a target bank")
+    p_prom.add_argument("project", help="Project directory")
+    p_prom.add_argument("--target-bank", required=True, help="Production bank ID")
+    p_prom.add_argument("--backfill", action="store_true", help="Trigger consolidation backfill")
+    p_prom.set_defaults(func=cmd_promote)
+
+    # ui
+    p_ui = sub.add_parser("ui", help="Open the project browser UI")
+    p_ui.add_argument("project", help="Project directory")
+    p_ui.set_defaults(func=cmd_ui)
+
+    args = parser.parse_args()
+    try:
+        args.func(args)
+    except KeyboardInterrupt:
+        console.print("\n[yellow]Interrupted.[/yellow]")
+        sys.exit(1)
+    except Exception as e:
+        console.print(f"[red]Error: {e}[/red]")
+        sys.exit(1)

hindsight-dev/hindsight_dev/mission_sandbox/helpers.py

@@ -0,0 +1,52 @@
+"""Shared helpers for mission-sandbox commands."""
+
+from __future__ import annotations
+
+import asyncio
+import time
+
+from rich.console import Console
+
+from .project import ObservationSample
+
+console = Console()
+
+
+def _run(coro):
+    return asyncio.get_event_loop().run_until_complete(coro)
+
+
+def wait_for_consolidation(client, bank_id: str, timeout: int = 600) -> None:
+    """Poll bank stats until no pending consolidation remains."""
+    start = time.time()
+    while time.time() - start < timeout:
+        stats = _run(client.banks.get_agent_stats(bank_id))
+        pending = getattr(stats, "pending_consolidation", 0)
+        if pending == 0:
+            return
+        console.print(f"  Waiting for consolidation... ({pending} pending)")
+        time.sleep(3)
+    raise TimeoutError(f"Consolidation did not complete within {timeout}s")
+
+
+def fetch_observations(client, bank_id: str) -> list[ObservationSample]:
+    """Fetch all observations from a bank."""
+    observations: list[ObservationSample] = []
+    offset = 0
+    while True:
+        page = client.list_memories(bank_id=bank_id, type="observation", limit=100, offset=offset)
+        items = page.items if hasattr(page, "items") else page.memories if hasattr(page, "memories") else []
+        if not items:
+            break
+        for mem in items:
+            mem_id = mem["id"] if isinstance(mem, dict) else mem.id
+            mem_text = mem["text"] if isinstance(mem, dict) else mem.text
+            observations.append(
+                ObservationSample(
+                    id=str(mem_id),
+                    text=mem_text,
+                    source_facts=[],
+                )
+            )
+        offset += len(items)
+    return observations

hindsight-dev/hindsight_dev/mission_sandbox/init.py

@@ -0,0 +1,65 @@
+"""init command: create project, ingest documents, consolidate, export observations."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from .helpers import _run, console, fetch_observations, wait_for_consolidation
+from .project import InitStep, Project
+
+
+def _collect_documents(path: Path) -> list[Path]:
+    """Collect .txt and .md files from a path (file or directory)."""
+    if path.is_file():
+        return [path]
+    if path.is_dir():
+        return sorted(f for f in path.rglob("*") if f.is_file() and f.suffix in (".txt", ".md"))
+    raise FileNotFoundError(f"Path not found: {path}")
+
+
+def run_init(
+    project_path: Path,
+    documents_path: Path,
+    bank_id: str,
+    api_url: str,
+) -> None:
+    from hindsight_client import Hindsight
+
+    # 1. Create project
+    proj = Project.create(project_path, bank_id=bank_id, api_url=api_url)
+    console.print(f"[bold]Created project at [cyan]{project_path}[/cyan][/bold]")
+
+    client = Hindsight(base_url=api_url)
+
+    # 2. Create bank (no mission for baseline)
+    console.print(f"[bold]Creating bank [cyan]{bank_id}[/cyan]...[/bold]")
+    client.create_bank(bank_id=bank_id, enable_observations=True)
+
+    # 3. Ingest documents
+    docs = _collect_documents(documents_path)
+    console.print(f"[bold]Ingesting {len(docs)} document(s)...[/bold]")
+    doc_names = []
+    for doc in docs:
+        content = doc.read_text()
+        console.print(f"  Retaining: {doc.name} ({len(content)} chars)")
+        client.retain(bank_id=bank_id, content=content, document_id=doc.stem)
+        doc_names.append(doc.name)
+
+    # 4. Trigger consolidation and wait
+    console.print("[bold]Triggering consolidation...[/bold]")
+    _run(client.banks.trigger_consolidation(bank_id))
+    wait_for_consolidation(client, bank_id)
+    console.print("[green]Consolidation complete.[/green]")
+
+    # 5. Export observations
+    console.print("[bold]Exporting observations...[/bold]")
+    observations = fetch_observations(client, bank_id)
+
+    # 6. Save to project
+    proj.observations = observations
+    step = InitStep(documents=doc_names, observations=observations)
+    step_path = proj.add_step(step)
+
+    console.print(f"[green bold]Done![/green bold] {len(observations)} observations")
+    console.print(f"  Project: {project_path}")
+    console.print(f"  History: {step_path.name}")

hindsight-dev/hindsight_dev/mission_sandbox/optimizer.py

@@ -0,0 +1,100 @@
+"""optimize command: generate improved mission from labeled observations."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from rich.panel import Panel
+
+from .helpers import console
+from .project import OptimizeStep, Project
+
+_OPTIMIZE_SYSTEM_PROMPT = """\
+You are an expert at writing observation missions for a memory system.
+
+An "observation mission" is a prompt that controls how raw facts get consolidated
+into observations. Observations are synthesized summaries derived from multiple raw facts.
+
+The user has labeled a set of observations as "good" (valuable) or "bad" (noise/unwanted).
+Each observation has a reason explaining why it was labeled that way.
+
+Your job: write an improved observation mission prompt that would produce more of the
+good observations and fewer of the bad ones.
+
+Rules:
+- The mission should be concise and actionable (a few sentences to a short paragraph)
+- Focus on what TO track and what NOT to track
+- Be specific about the domain based on the patterns you see
+- If there's an existing mission, improve it; otherwise write one from scratch
+
+Respond with ONLY the mission text, no explanation or wrapper.
+"""
+
+
+def run_optimize(
+    project_path: Path,
+    model: str,
+) -> None:
+    import litellm
+
+    proj = Project.load(project_path)
+
+    labeled = [o for o in proj.observations if o.label is not None]
+    if not labeled:
+        console.print("[red]No labeled observations found. Run agent-label first.[/red]")
+        return
+
+    good = [o for o in labeled if o.label == "good"]
+    bad = [o for o in labeled if o.label == "bad"]
+    console.print(f"[bold]Optimizing mission from {len(good)} good + {len(bad)} bad examples...[/bold]")
+
+    # Build the examples section
+    examples_parts: list[str] = []
+    for o in good:
+        examples_parts.append(f"GOOD: {o.text}\n  Reason: {o.reason}")
+    for o in bad:
+        examples_parts.append(f"BAD: {o.text}\n  Reason: {o.reason}")
+    examples_str = "\n\n".join(examples_parts)
+
+    current_mission = proj.mission or "(no mission set — using system defaults)"
+
+    user_msg = f"""\
+## Current mission
+{current_mission}
+
+## Labeled observations
+{examples_str}
+
+Write an improved observation mission.
+"""
+
+    response = litellm.completion(
+        model=model,
+        messages=[
+            {"role": "system", "content": _OPTIMIZE_SYSTEM_PROMPT},
+            {"role": "user", "content": user_msg},
+        ],
+        temperature=0.3,
+    )
+    new_mission = response.choices[0].message.content.strip()
+
+    console.print("\n[bold]Proposed mission:[/bold]")
+    console.print(Panel(new_mission, title="New Mission", border_style="green"))
+
+    if proj.mission:
+        console.print("\n[bold]Previous mission:[/bold]")
+        console.print(Panel(proj.mission, title="Old Mission", border_style="dim"))
+
+    # Record history
+    step = OptimizeStep(
+        model=model,
+        previous_mission=proj.mission,
+        proposed_mission=new_mission,
+        good_count=len(good),
+        bad_count=len(bad),
+    )
+    proj.mission = new_mission
+    step_path = proj.add_step(step)
+
+    console.print("\n[green bold]Mission saved.[/green bold]")
+    console.print(f"  History: {step_path.name}")

hindsight-dev/hindsight_dev/mission_sandbox/project.py

@@ -0,0 +1,189 @@
+"""Project model: a directory-based project with full history tracking.
+
+Project structure:
+    my-project/
+        project.json        # metadata: bank_id, api_url, current mission
+        history/
+            0001_init.json
+            0002_label.json
+            0003_optimize.json
+            0004_run.json
+            ...
+"""
+
+from __future__ import annotations
+
+import json
+from dataclasses import asdict, dataclass, field
+from datetime import datetime, timezone
+from pathlib import Path
+
+
+@dataclass
+class ObservationSample:
+    id: str
+    text: str
+    source_facts: list[str]
+    label: str | None = None
+    reason: str | None = None
+
+
+# -- History step types --------------------------------------------------------
+
+
+@dataclass
+class InitStep:
+    """Recorded when `init` ingests documents and runs first consolidation."""
+
+    type: str = "init"
+    timestamp: str = ""
+    documents: list[str] = field(default_factory=list)
+    observations: list[ObservationSample] = field(default_factory=list)
+
+
+@dataclass
+class LabelStep:
+    """Recorded when `agent-label` scores observations."""
+
+    type: str = "label"
+    timestamp: str = ""
+    instructions: str = ""
+    model: str = ""
+    observations: list[ObservationSample] = field(default_factory=list)
+
+
+@dataclass
+class OptimizeStep:
+    """Recorded when `optimize` proposes a new mission."""
+
+    type: str = "optimize"
+    timestamp: str = ""
+    model: str = ""
+    previous_mission: str | None = None
+    proposed_mission: str = ""
+    good_count: int = 0
+    bad_count: int = 0
+
+
+@dataclass
+class RunStep:
+    """Recorded when `run` applies mission, re-consolidates, exports new observations."""
+
+    type: str = "run"
+    timestamp: str = ""
+    mission_applied: str = ""
+    observations: list[ObservationSample] = field(default_factory=list)
+
+
+@dataclass
+class PromoteStep:
+    """Recorded when `promote` pushes mission to a production bank."""
+
+    type: str = "promote"
+    timestamp: str = ""
+    target_bank: str = ""
+    mission: str = ""
+    backfill: bool = False
+
+
+STEP_TYPES = {
+    "init": InitStep,
+    "label": LabelStep,
+    "optimize": OptimizeStep,
+    "run": RunStep,
+    "promote": PromoteStep,
+}
+
+
+def _now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+# -- Project -------------------------------------------------------------------
+
+
+@dataclass
+class Project:
+    bank_id: str
+    api_url: str
+    mission: str | None = None
+    created_at: str = ""
+    observations: list[ObservationSample] = field(default_factory=list)
+
+    _path: Path | None = field(default=None, repr=False)
+
+    @classmethod
+    def create(cls, path: Path, bank_id: str, api_url: str) -> Project:
+        """Create a new project directory."""
+        path.mkdir(parents=True, exist_ok=True)
+        (path / "history").mkdir(exist_ok=True)
+        proj = cls(bank_id=bank_id, api_url=api_url, created_at=_now(), _path=path)
+        proj._save_meta()
+        return proj
+
+    @classmethod
+    def load(cls, path: Path) -> Project:
+        """Load an existing project."""
+        meta_file = path / "project.json"
+        if not meta_file.exists():
+            raise FileNotFoundError(f"No project found at {path} (missing project.json)")
+        raw = json.loads(meta_file.read_text())
+        observations = [ObservationSample(**o) for o in raw.get("observations", [])]
+        proj = cls(
+            bank_id=raw["bank_id"],
+            api_url=raw["api_url"],
+            mission=raw.get("mission"),
+            created_at=raw.get("created_at", ""),
+            observations=observations,
+            _path=path,
+        )
+        return proj
+
+    @property
+    def path(self) -> Path:
+        assert self._path is not None
+        return self._path
+
+    def _save_meta(self) -> None:
+        """Write project.json."""
+        data = {
+            "bank_id": self.bank_id,
+            "api_url": self.api_url,
+            "mission": self.mission,
+            "created_at": self.created_at,
+            "observations": [asdict(o) for o in self.observations],
+        }
+        (self.path / "project.json").write_text(json.dumps(data, indent=2))
+
+    def save(self) -> None:
+        self._save_meta()
+
+    # -- History ---------------------------------------------------------------
+
+    def _next_step_number(self) -> int:
+        history_dir = self.path / "history"
+        existing = sorted(history_dir.glob("*.json"))
+        if not existing:
+            return 1
+        last = existing[-1].stem  # e.g. "0003_optimize"
+        return int(last.split("_")[0]) + 1
+
+    def add_step(self, step: InitStep | LabelStep | OptimizeStep | RunStep | PromoteStep) -> Path:
+        """Append a history step and save project metadata."""
+        step.timestamp = _now()
+        num = self._next_step_number()
+        filename = f"{num:04d}_{step.type}.json"
+        step_path = self.path / "history" / filename
+        step_path.write_text(json.dumps(asdict(step), indent=2))
+        self._save_meta()
+        return step_path
+
+    def list_steps(self) -> list[dict]:
+        """Load all history steps in order."""
+        history_dir = self.path / "history"
+        steps = []
+        for f in sorted(history_dir.glob("*.json")):
+            raw = json.loads(f.read_text())
+            raw["_file"] = f.name
+            steps.append(raw)
+        return steps

hindsight-dev/hindsight_dev/mission_sandbox/promote.py

@@ -0,0 +1,43 @@
+"""promote command: push optimized mission to a target bank."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from .helpers import _run, console
+from .project import Project, PromoteStep
+
+
+def run_promote(
+    project_path: Path,
+    target_bank: str,
+    backfill: bool = False,
+) -> None:
+    from hindsight_client import Hindsight
+
+    proj = Project.load(project_path)
+
+    if not proj.mission:
+        console.print("[red]No mission in project. Run optimize first.[/red]")
+        return
+
+    client = Hindsight(base_url=proj.api_url)
+
+    console.print(f"[bold]Promoting mission to bank [cyan]{target_bank}[/cyan]...[/bold]")
+    client.update_bank_config(bank_id=target_bank, observations_mission=proj.mission)
+    console.print("[green]Mission updated.[/green]")
+
+    if backfill:
+        console.print("[bold]Triggering backfill consolidation...[/bold]")
+        _run(client.banks.clear_observations(target_bank))
+        _run(client.banks.trigger_consolidation(target_bank))
+        console.print("[green]Backfill consolidation triggered.[/green]")
+
+    # Record history
+    step = PromoteStep(
+        target_bank=target_bank,
+        mission=proj.mission,
+        backfill=backfill,
+    )
+    step_path = proj.add_step(step)
+    console.print(f"  History: {step_path.name}")

hindsight-dev/hindsight_dev/mission_sandbox/run.py

@@ -0,0 +1,54 @@
+"""run command: apply mission, re-consolidate, update project with new observations."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from .helpers import _run, console, fetch_observations, wait_for_consolidation
+from .project import Project, RunStep
+
+
+def run_run(project_path: Path) -> None:
+    from hindsight_client import Hindsight
+
+    proj = Project.load(project_path)
+
+    if not proj.mission:
+        console.print("[red]No mission in project. Run optimize first.[/red]")
+        return
+
+    client = Hindsight(base_url=proj.api_url)
+
+    # 1. Update observation mission on the sandbox bank
+    console.print(f"[bold]Updating mission on bank [cyan]{proj.bank_id}[/cyan]...[/bold]")
+    client.update_bank_config(bank_id=proj.bank_id, observations_mission=proj.mission)
+
+    # 2. Clear existing observations and re-consolidate
+    console.print("[bold]Clearing existing observations...[/bold]")
+    _run(client.banks.clear_observations(proj.bank_id))
+
+    console.print("[bold]Re-running consolidation with new mission...[/bold]")
+    _run(client.banks.trigger_consolidation(proj.bank_id))
+    wait_for_consolidation(client, proj.bank_id)
+    console.print("[green]Consolidation complete.[/green]")
+
+    # 3. Fetch new observations
+    new_observations = fetch_observations(client, proj.bank_id)
+
+    # Show summary
+    old_count = len(proj.observations)
+    old_labeled = sum(1 for o in proj.observations if o.label is not None)
+    console.print(f"\n[bold]Previous:[/bold] {old_count} observations ({old_labeled} labeled)")
+    console.print(f"[bold]New:[/bold] {len(new_observations)} observations (all unlabeled, ready for agent-label)")
+
+    # 4. Record history and update project
+    step = RunStep(
+        mission_applied=proj.mission,
+        observations=new_observations,
+    )
+    proj.observations = new_observations
+    step_path = proj.add_step(step)
+
+    console.print(f"\n[green bold]Done![/green bold] Dataset updated with {len(new_observations)} observations.")
+    console.print(f"  History: {step_path.name}")
+    console.print("[dim]Run agent-label next to score these observations.[/dim]")

hindsight-dev/hindsight_dev/mission_sandbox/ui.py

@@ -0,0 +1,221 @@
+"""Streamlit UI for browsing mission-sandbox projects."""
+
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+
+import streamlit as st
+
+# Project path is passed as CLI arg by the `mission-sandbox ui` command
+if len(sys.argv) > 1:
+    PROJECT_PATH = Path(sys.argv[1])
+else:
+    PROJECT_PATH = None
+
+
+def load_project(path: Path) -> dict:
+    meta_file = path / "project.json"
+    if not meta_file.exists():
+        return {}
+    return json.loads(meta_file.read_text())
+
+
+def load_steps(path: Path) -> list[dict]:
+    history_dir = path / "history"
+    if not history_dir.exists():
+        return []
+    steps = []
+    for f in sorted(history_dir.glob("*.json")):
+        raw = json.loads(f.read_text())
+        raw["_file"] = f.name
+        steps.append(raw)
+    return steps
+
+
+def render_step_init(step: dict) -> None:
+    docs = step.get("documents", [])
+    obs = step.get("observations", [])
+    st.write(f"**Documents ingested:** {len(docs)}")
+    if docs:
+        with st.expander(f"Documents ({len(docs)})"):
+            for d in docs:
+                st.code(d, language=None)
+    st.write(f"**Observations produced:** {len(obs)}")
+    if obs:
+        with st.expander(f"Observations ({len(obs)})"):
+            for o in obs:
+                st.markdown(f"- {o['text']}")
+
+
+def render_step_label(step: dict) -> None:
+    st.write(f"**Model:** `{step.get('model', '?')}`")
+    st.write("**Instructions:**")
+    st.info(step.get("instructions", ""))
+    obs = step.get("observations", [])
+    good = [o for o in obs if o.get("label") == "good"]
+    bad = [o for o in obs if o.get("label") == "bad"]
+    st.write(f"**Results:** {len(good)} good, {len(bad)} bad out of {len(obs)} total")
+
+    col1, col2 = st.columns(2)
+    with col1:
+        st.markdown("##### Good")
+        for o in good:
+            with st.container(border=True):
+                st.markdown(o["text"])
+                st.caption(o.get("reason", ""))
+    with col2:
+        st.markdown("##### Bad")
+        for o in bad:
+            with st.container(border=True):
+                st.markdown(o["text"])
+                st.caption(o.get("reason", ""))
+
+
+def render_step_optimize(step: dict) -> None:
+    st.write(f"**Model:** `{step.get('model', '?')}`")
+    st.write(f"**Input:** {step.get('good_count', 0)} good, {step.get('bad_count', 0)} bad examples")
+
+    prev = step.get("previous_mission")
+    proposed = step.get("proposed_mission", "")
+
+    if prev:
+        st.write("**Previous mission:**")
+        st.warning(prev)
+    else:
+        st.write("**Previous mission:** _(none -- baseline)_")
+
+    st.write("**Proposed mission:**")
+    st.success(proposed)
+
+
+def render_step_run(step: dict) -> None:
+    mission = step.get("mission_applied", "")
+    obs = step.get("observations", [])
+    st.write("**Mission applied:**")
+    st.info(mission)
+    st.write(f"**Observations produced:** {len(obs)}")
+    if obs:
+        with st.expander(f"Observations ({len(obs)})"):
+            for o in obs:
+                st.markdown(f"- {o['text']}")
+
+
+def render_step_promote(step: dict) -> None:
+    st.write(f"**Target bank:** `{step.get('target_bank', '?')}`")
+    st.write(f"**Backfill:** {'Yes' if step.get('backfill') else 'No'}")
+    st.write("**Mission:**")
+    st.success(step.get("mission", ""))
+
+
+STEP_RENDERERS = {
+    "init": render_step_init,
+    "label": render_step_label,
+    "optimize": render_step_optimize,
+    "run": render_step_run,
+    "promote": render_step_promote,
+}
+
+STEP_ICONS = {
+    "init": ":inbox_tray:",
+    "label": ":label:",
+    "optimize": ":sparkles:",
+    "run": ":arrows_counterclockwise:",
+    "promote": ":rocket:",
+}
+
+
+def main() -> None:
+    st.set_page_config(page_title="Mission Sandbox", page_icon=":microscope:", layout="wide")
+    st.title(":microscope: Mission Sandbox")
+
+    if PROJECT_PATH is None:
+        st.error("No project path provided. Run: `mission-sandbox ui <project-dir>`")
+        return
+
+    project = load_project(PROJECT_PATH)
+    if not project:
+        st.error(f"No project found at `{PROJECT_PATH}`")
+        return
+
+    steps = load_steps(PROJECT_PATH)
+
+    # -- Sidebar: project info -------------------------------------------------
+    with st.sidebar:
+        st.header("Project")
+        st.write(f"**Path:** `{PROJECT_PATH}`")
+        st.write(f"**Bank:** `{project.get('bank_id', '?')}`")
+        st.write(f"**API:** `{project.get('api_url', '?')}`")
+        st.write(f"**Created:** {project.get('created_at', '?')[:19]}")
+
+        current_mission = project.get("mission")
+        if current_mission:
+            st.divider()
+            st.subheader("Current Mission")
+            st.info(current_mission)
+
+        current_obs = project.get("observations", [])
+        if current_obs:
+            st.divider()
+            st.subheader("Current Observations")
+            good = sum(1 for o in current_obs if o.get("label") == "good")
+            bad = sum(1 for o in current_obs if o.get("label") == "bad")
+            unlabeled = sum(1 for o in current_obs if o.get("label") is None)
+            st.write(f"Total: **{len(current_obs)}**")
+            if good or bad:
+                st.write(f":white_check_mark: {good} good  :x: {bad} bad  :grey_question: {unlabeled} unlabeled")
+
+        st.divider()
+        st.subheader("Steps")
+        st.write(f"**{len(steps)}** history entries")
+
+    # -- Main: history timeline ------------------------------------------------
+    if not steps:
+        st.info("No history yet. Run `mission-sandbox init` to get started.")
+        return
+
+    # Summary metrics across rounds
+    label_rounds = [s for s in steps if s["type"] == "label"]
+    optimize_rounds = [s for s in steps if s["type"] == "optimize"]
+    if label_rounds:
+        col1, col2, col3 = st.columns(3)
+        with col1:
+            st.metric("Label rounds", len(label_rounds))
+        with col2:
+            st.metric("Optimize rounds", len(optimize_rounds))
+        with col3:
+            # Show improvement: good ratio in first vs last label round
+            first_label = label_rounds[0]
+            last_label = label_rounds[-1]
+            first_obs = first_label.get("observations", [])
+            last_obs = last_label.get("observations", [])
+            first_good = sum(1 for o in first_obs if o.get("label") == "good") / max(len(first_obs), 1)
+            last_good = sum(1 for o in last_obs if o.get("label") == "good") / max(len(last_obs), 1)
+            st.metric(
+                "Good ratio",
+                f"{last_good:.0%}",
+                delta=f"{last_good - first_good:+.0%}" if len(label_rounds) > 1 else None,
+            )
+
+    st.divider()
+
+    # Render each step
+    for i, step in enumerate(steps):
+        step_type = step.get("type", "unknown")
+        icon = STEP_ICONS.get(step_type, ":question:")
+        timestamp = step.get("timestamp", "")[:19].replace("T", " ")
+        filename = step.get("_file", "")
+
+        with st.expander(
+            f"{icon} **Step {i + 1}: {step_type.upper()}** — {timestamp}  `{filename}`", expanded=(i == len(steps) - 1)
+        ):
+            renderer = STEP_RENDERERS.get(step_type)
+            if renderer:
+                renderer(step)
+            else:
+                st.json(step)
+
+
+if __name__ == "__main__":
+    main()

hindsight-dev/pyproject.toml

@@ -15,6 +15,8 @@ dependencies = [
     "rich>=13.0.0",
     "pydantic>=2.0.0",
     "httpx>=0.27.0",
+    "litellm>=1.0.0",
+    "hindsight-client>=0.7.0",
 ]
 
 [project.optional-dependencies]
@@ -29,6 +31,7 @@ packages = ["hindsight_dev", "benchmarks", "upgrade_tests"]
 
 [tool.uv.sources]
 hindsight-api = { workspace = true }
+hindsight-client = { workspace = true }
 
 [project.scripts]
 generate-openapi = "hindsight_dev.generate_openapi:generate_openapi_spec"
@@ -40,6 +43,7 @@ check-openapi-compatibility = "hindsight_dev.check_openapi_compatibility:main"
 cli-coverage-check = "hindsight_dev.cli_coverage_check:main"
 client-coverage-check = "hindsight_dev.client_coverage_check:main"
 perf-test = "benchmarks.perf.system_perf:main"
+mission-sandbox = "hindsight_dev.mission_sandbox.cli:main"
 
 [dependency-groups]
 dev = [