Turn the Conversation into Your Own Repo

1Use my chapter-one brief and selected current files for chapter 2 only.2Propose a small setup/manifest/managed-search increment; wait for approval3before applying it. Preserve existing files and the selected corpus identity.4Check setup files, checksum checks, persisted remote IDs before polling,5and resume behavior. Cite current official API contracts; do not invent fields.6Use strict offline tests first; inspect actual responses only after I approve a live run.7Do not read secrets, commit/push, provision resources, or make paid calls8without my explicit approval. Do not jump to later chapters.9Explain each change; debug only with redacted errors; I review the diff and commit.

1Review chapter 2's proposed diff and offline test evidence only.2Check setup order, gitignored secrets/media, checksum integrity, saved IDs,3and processing/failed/interrupted resume cases against current official API contracts.4Separate offline evidence from actual provider responses; invent neither.5Flag unexplained changes and redact errors. Do not read secrets or jump chapters.6Do not commit/push, provision, or make paid calls without my explicit approval.7Show what I must inspect in git diff and git diff --cached; I make the commit.

1Here is a brief from a video AI experiment I ran: [paste your chapter-one brief].2 3Propose a Python project layout for a small lab that:41. builds a media manifest (checksums, durations, handwritten ground truth),52. uploads three local clips to TwelveLabs and indexes them with Marengo 3.06   (model_options visual + audio),73. runs five fixed search queries and saves results locally for grading.8 9Constraints: API key from a gitignored .env via python-dotenv; every remote ID10persisted to a gitignored state.json BEFORE polling, so a crash resumes instead11of duplicating; no secrets or media in Git. Plan first — no code yet.

1twelvelabs-video-lab/2  README.md               # layout, setup, run order, Git discipline3  requirements.txt        # twelvelabs, python-dotenv, numpy, boto34  .env.example            # placeholder names only5  .gitignore              # .env, .venv, media/, local-results/, vectors/, *state.json6  media-manifest.json     # generated, then ground_truth filled in by hand7  media/                  # your three clips (gitignored)8  local-results/          # model outputs + your grading (gitignored)9  state.json              # remote IDs for resume (gitignored)10  media_manifest.py       # this chapter11  managed_search.py       # this chapter

1python3 --version    # 3.10 or newer2git --version3ffmpeg -version      # first line shows the version4ffprobe -version5node --version       # chapter 6 only; v20 or newer

1git clone https://github.com/YOUR-USERNAME/twelvelabs-video-lab.git2cd twelvelabs-video-lab3pwd   # confirm you are inside your own repository

1twelvelabs2python-dotenv3numpy4boto3

1# Placeholders only. Real values stay in your local, untracked .env.2TWELVELABS_API_KEY=tlk_your_key_here3AWS_PROFILE=twelvelabs-video-lab4AWS_REGION=us-east-15LAB_BUCKET=your-unique-bucket-name6LAB_ACCOUNT_ID=123456789012

1# Secrets and local state never enter Git.2.env3.venv/4__pycache__/5# Local results, media, vectors and resume state stay on your disk.6local-results/7media/8vectors/9state.json10bedrock-state.json11# The workbench builds its fixture from local results; only synthetic12# demo fixtures under workbench/public/demo/ are committed.13workbench/public/fixtures.json14workbench/node_modules/15workbench/dist/

1# twelvelabs-video-lab2 3My own video AI lab: Jockey -> managed search -> Pegasus -> Marengo -> Bedrock.4 5## Setup6python3 -m venv .venv && source .venv/bin/activate7pip install -r requirements.txt8cp .env.example .env   # add my own TwelveLabs key locally; never commit .env9 10## Rules11- Only my own recordings go in media/ (gitignored).12- Remote IDs live in state.json (gitignored) so runs resume.13- Run git status and git diff --cached before every commit.

1git status --short2git add .gitignore README.md requirements.txt .env.example3git diff --cached --stat4git diff --cached5# Proceed only if exactly the four setup files are staged, with placeholders only.6git commit -m "Lab skeleton: no secrets, no media"7git log -1 --oneline8git status --short

1python3 -m venv .venv && source .venv/bin/activate2pip install -r requirements.txt3cp .env.example .env4# Edit .env locally: replace only the TwelveLabs key placeholder.5# Leave AWS placeholders alone until chapter five.6git check-ignore .env .venv/7git status --short

1python -m py_compile media_manifest.py managed_search.py2git status --short3# The two new Python files should be visible; secrets and videos should not be.

1mkdir -p media2# Copy your three .mp4 files into this folder using your file manager.3# Keep them directly inside media/, not inside a subfolder.4python - <<'PY'5from pathlib import Path6files = sorted(Path("media").glob("*.mp4"))7for file in files:8    print(file)9print(f"{len(files)} MP4 files")10assert len(files) == 3, "Put your three selected MP4 clips directly in media/"11PY12git check-ignore media/

1"""Build media-manifest.json: duration, checksum, ground truth per clip."""2import hashlib3import json4import subprocess5from pathlib import Path6 7ROOT = Path(__file__).parent8MEDIA = ROOT / "media"9 10 11def sha256(path: Path) -> str:12    h = hashlib.sha256()13    with path.open("rb") as f:14        for chunk in iter(lambda: f.read(1 << 20), b""):15            h.update(chunk)16    return h.hexdigest()17 18 19def duration_sec(path: Path) -> float:20    out = subprocess.run(21        ["ffprobe", "-v", "error", "-show_entries", "format=duration",22         "-of", "default=noprint_wrappers=1:nokey=1", str(path)],23        capture_output=True, text=True, check=True,24    )25    return round(float(out.stdout.strip()), 2)26 27 28def main() -> None:29    clips = []30    for path in sorted(MEDIA.glob("*.mp4")):31        clips.append({32            "id": path.stem,33            "file": f"media/{path.name}",34            "sha256": sha256(path),35            "duration_sec": duration_sec(path),36            "ground_truth": [],  # fill in by hand — this is your grading truth37        })38    out = ROOT / "media-manifest.json"39    out.write_text(json.dumps({"corpus": "desk-corpus", "clips": clips}, indent=2))40    print(f"wrote {out} with {len(clips)} clips — now fill in ground_truth")41 42 43if __name__ == "__main__":44    main()

1python media_manifest.py2python -m json.tool media-manifest.json3# The generator must report 3 clips before you proceed.

1"ground_truth": [2  {3    "query": "a person opens a blue notebook",4    "start_sec": 2.0,5    "end_sec": 4.0,6    "observation": "The notebook opens in this interval; checked by replaying."7  }8]

1"""Chapter 2: Marengo 3.0 managed search — upload, index, query.2 3Documentation-checked against the TwelveLabs Python SDK search reference;4live validation pending. Contract notes:5- indexes.create(index_name=..., models=[{"model_name": "marengo3.0",6  "model_options": ["visual", "audio"]}])7- client.search.query(index_id=..., query_text=..., search_options=["visual", "audio"])8- Result fields: rank, video_id, start/end timestamps, optional user_metadata.9  There is no confidence field; rank order is the signal.10"""11import json12import os13import time14from pathlib import Path15 16from dotenv import load_dotenv17from twelvelabs import TwelveLabs18 19load_dotenv()20client = TwelveLabs(api_key=os.environ["TWELVELABS_API_KEY"])21 22ROOT = Path(__file__).parent23STATE_PATH = ROOT / "state.json"24RESULTS = ROOT / "local-results"25RESULTS.mkdir(exist_ok=True)26 27state = json.loads(STATE_PATH.read_text()) if STATE_PATH.exists() else {}28 29 30def save_state() -> None:31    STATE_PATH.write_text(json.dumps(state, indent=2))32 33 34def verify_checksums(manifest: dict) -> None:35    """Never reuse a stored asset ID for a file that changed on disk."""36    import hashlib37    for clip in manifest["clips"]:38        h = hashlib.sha256((ROOT / clip["file"]).read_bytes()).hexdigest()39        if h != clip["sha256"]:40            raise RuntimeError(f"{clip['id']}: file changed since manifest — re-record or re-manifest")41 42 43def ensure_index() -> str:44    if "index_id" not in state:45        index = client.indexes.create(46            index_name="twelvelabs-video-lab",47            models=[{"model_name": "marengo3.0", "model_options": ["visual", "audio"]}],48        )49        state["index_id"] = index.id50        save_state()51    return state["index_id"]52 53 54def ensure_assets(manifest: dict) -> None:55    """Upload clips; revalidate saved IDs instead of blindly skipping."""56    state.setdefault("assets", {})57    for clip in manifest["clips"]:58        clip_id = clip["id"]59        if clip_id in state["assets"]:60            saved = state["assets"][clip_id]61            if saved.get("sha256") != clip["sha256"]:62                raise RuntimeError(f"{clip_id}: checksum mismatch — delete the stale asset and re-run")63            asset = client.assets.retrieve(saved["asset_id"])64            if asset.status == "failed":65                raise RuntimeError(f"{clip_id}: asset {saved['asset_id']} failed processing — re-upload")66            print(f"skip {clip_id}: asset {saved['asset_id']} status={asset.status}")67            continue68        with (ROOT / clip["file"]).open("rb") as f:69            asset = client.assets.create(method="direct", file=f)70        state["assets"][clip_id] = {"asset_id": asset.id, "sha256": clip["sha256"]}71        save_state()72        print(f"uploaded {clip_id} -> {asset.id}")73 74 75def wait_asset(api, asset_id: str, timeout_sec: int = 1800):76    """Poll a persisted upload; timeout preserves the ID for the next run."""77    deadline = time.monotonic() + timeout_sec78    while time.monotonic() < deadline:79        asset = api.assets.retrieve(asset_id)80        if asset.status == "ready":81            return asset82        if asset.status in ("failed", "canceled"):83            raise RuntimeError(f"asset {asset_id}: {asset.status}")84        time.sleep(5)85    raise TimeoutError(f"asset {asset_id} still processing — resume the same ID")86 87 88def wait_indexed(index_id: str, timeout_sec: int = 1800) -> None:89    """Create actual index entries after upload readiness; persist and resume IDs."""90    for clip_id, saved in state["assets"].items():91        wait_asset(client, saved["asset_id"], timeout_sec)92        if "indexed_asset_id" not in saved:93            entry = client.indexes.indexed_assets.create(index_id, asset_id=saved["asset_id"])94            saved["indexed_asset_id"] = entry.id95            save_state()96        deadline = time.monotonic() + timeout_sec97        while time.monotonic() < deadline:98            entry = client.indexes.indexed_assets.retrieve(index_id, saved["indexed_asset_id"])99            if entry.status == "ready":100                print(f"indexed: {clip_id}")101                break102            if entry.status in ("failed", "canceled"):103                raise RuntimeError(f"indexed asset {saved['indexed_asset_id']}: {entry.status}")104            time.sleep(5)105        else:106            raise TimeoutError(f"{clip_id}: still indexing — resume indexed_asset_id")107 108 109def search(index_id: str, query: str) -> list[dict]:110    """Managed search over both modalities. Fields per SDK reference:111    rank, video_id, start/end timestamps, optional user_metadata."""112    result = client.search.query(113        index_id=index_id,114        query_text=query,115        search_options=["visual", "audio"],116    )117    hits = []118    for item in result:119        hits.append({120            "rank": item.rank,121            "video_id": item.video_id,122            "start": item.start,123            "end": item.end,124            "user_metadata": getattr(item, "user_metadata", None),125        })126    return hits127 128 129QUERIES = [130    "a person opens a blue notebook",131    "the red mug is moved across the desk",132    "someone claps once",133    "a pen rolls off the table",134    "a person opens a window",  # absent on purpose — expect no good hit135]136 137 138def main() -> None:139    manifest = json.loads((ROOT / "media-manifest.json").read_text())140    verify_checksums(manifest)141    index_id = ensure_index()142    ensure_assets(manifest)143    wait_indexed(index_id)144    out = {}145    for q in QUERIES:146        out[q] = search(index_id, q)147        print(f"{q!r}: {len(out[q])} hits, top={out[q][0] if out[q] else None}")148    (RESULTS / "managed-search.json").write_text(json.dumps(out, indent=2))149    print("saved local-results/managed-search.json — grade it against your ground truth")150 151 152if __name__ == "__main__":153    main()

1python managed_search.py2# On success, inspect local-results/managed-search.json and state.json locally.3# Neither belongs in a commit. Do not share credentials or unredacted local state.

1git status --short2git diff                         # inspect changes to already tracked files3git add README.md requirements.txt .env.example .gitignore \4        media-manifest.json media_manifest.py managed_search.py5git diff --cached --stat6git diff --cached                # inspect exactly what the commit will contain7# Stop if secrets, videos, results or state are staged.8git commit -m "Chapter 2: manifest and managed Marengo 3.0 search"9git status --short10# Optional, only when you decide to send this reviewed commit to your private remote:11# git push