diff --git a/README.md b/README.md index 3b47b38..a3d7e67 100644 --- a/README.md +++ b/README.md @@ -28,8 +28,15 @@ GAM by what is said in it. Then preview and render it: - text, lower thirds and pictures on screen, timed to the scenes and lines they belong to; - end credits from the library's attribution, and a description with chapters and credits. -A finished render can be saved back into GAM. AI drafting and the full storyboard -editor are next (M3, M4). +A finished render can be saved back into GAM. + +**The assistant (M4/M5).** A panel beside every page that knows the whole video format. +Describe a video and it drafts one from your GAM library, or ask for any change to the +one open. What it would change is listed for you to apply or discard. It runs in the +background, so a request finishes with the page closed. Set up its AI provider +(Anthropic, OpenAI, DeepSeek, Ollama or a custom endpoint) under Settings. + +The full storyboard editor is next (M3). ## Read first diff --git a/backend/alembic/versions/20261016_1016_add_ai_provider_and_usage_event_tables.py b/backend/alembic/versions/20261016_1016_add_ai_provider_and_usage_event_tables.py new file mode 100644 index 0000000..e57170c --- /dev/null +++ b/backend/alembic/versions/20261016_1016_add_ai_provider_and_usage_event_tables.py @@ -0,0 +1,87 @@ +"""Add the aiprovider and usageevent tables (M4/M5, the assistant's first half). + +Both are copied from GAM's, which came from gecko-notes. `usageevent.project_id` and +`turn_id` are plain indexed columns, not foreign keys: deleting a project must not +delete the record of what was spent on it. + +Revision ID: 4a8c2f6e9d13 +Revises: 7e2d5a91b3c6 +Create Date: 2026-10-16 10:16:48.213291+00:00 +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa +import sqlmodel + + +revision: str = '4a8c2f6e9d13' +down_revision: Union[str, None] = '7e2d5a91b3c6' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table('aiprovider', + sa.Column('id', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('user_id', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('name', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('provider_type', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('api_key', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('base_url', sqlmodel.sql.sqltypes.AutoString(), nullable=True), + sa.Column('model', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('max_tokens', sa.Integer(), nullable=False), + sa.Column('supports_images', sa.Boolean(), nullable=False), + sa.Column('use_anthropic_api', sa.Boolean(), nullable=False), + sa.Column('extra_params', sqlmodel.sql.sqltypes.AutoString(), nullable=True), + sa.Column('enabled', sa.Boolean(), nullable=False), + sa.Column('is_active', sa.Boolean(), nullable=False), + sa.Column('created_at', sa.DateTime(), nullable=False), + sa.Column('updated_at', sa.DateTime(), nullable=False), + sa.PrimaryKeyConstraint('id') + ) + with op.batch_alter_table('aiprovider', schema=None) as batch_op: + batch_op.create_index(batch_op.f('ix_aiprovider_user_id'), ['user_id'], unique=False) + + op.create_table('usageevent', + sa.Column('id', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('user_id', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('project_id', sqlmodel.sql.sqltypes.AutoString(), nullable=True), + sa.Column('kind', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('provider', sqlmodel.sql.sqltypes.AutoString(), nullable=True), + sa.Column('model', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('units', sa.Integer(), nullable=False), + sa.Column('unit_type', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('cost', sa.Float(), nullable=True), + sa.Column('currency', sqlmodel.sql.sqltypes.AutoString(), nullable=True), + sa.Column('cost_estimated', sa.Boolean(), nullable=True), + sa.Column('turn_id', sqlmodel.sql.sqltypes.AutoString(), nullable=True), + sa.Column('created_at', sa.DateTime(), nullable=False), + sa.PrimaryKeyConstraint('id') + ) + with op.batch_alter_table('usageevent', schema=None) as batch_op: + batch_op.create_index(batch_op.f('ix_usageevent_created_at'), ['created_at'], unique=False) + batch_op.create_index(batch_op.f('ix_usageevent_kind'), ['kind'], unique=False) + batch_op.create_index(batch_op.f('ix_usageevent_project_id'), ['project_id'], unique=False) + batch_op.create_index(batch_op.f('ix_usageevent_provider'), ['provider'], unique=False) + batch_op.create_index(batch_op.f('ix_usageevent_turn_id'), ['turn_id'], unique=False) + batch_op.create_index('ix_usageevent_user_created', ['user_id', 'created_at'], unique=False) + batch_op.create_index(batch_op.f('ix_usageevent_user_id'), ['user_id'], unique=False) + + + +def downgrade() -> None: + with op.batch_alter_table('usageevent', schema=None) as batch_op: + batch_op.drop_index(batch_op.f('ix_usageevent_user_id')) + batch_op.drop_index('ix_usageevent_user_created') + batch_op.drop_index(batch_op.f('ix_usageevent_turn_id')) + batch_op.drop_index(batch_op.f('ix_usageevent_provider')) + batch_op.drop_index(batch_op.f('ix_usageevent_project_id')) + batch_op.drop_index(batch_op.f('ix_usageevent_kind')) + batch_op.drop_index(batch_op.f('ix_usageevent_created_at')) + + op.drop_table('usageevent') + with op.batch_alter_table('aiprovider', schema=None) as batch_op: + batch_op.drop_index(batch_op.f('ix_aiprovider_user_id')) + + op.drop_table('aiprovider') diff --git a/backend/alembic/versions/20261016_1027_add_assistant_session_and_turn_tables.py b/backend/alembic/versions/20261016_1027_add_assistant_session_and_turn_tables.py new file mode 100644 index 0000000..e2997f7 --- /dev/null +++ b/backend/alembic/versions/20261016_1027_add_assistant_session_and_turn_tables.py @@ -0,0 +1,84 @@ +"""Add the assistant_session and assistant_turn tables (M4/M5). + +A turn is a job, so it carries the job fields the runner reads (status, stage, +progress, detail, error_message, updated_at) next to everything the turn is: its +prompt, the draft it started from, its steps, its reply and its proposal. + +Revision ID: 9b3e7d215f60 +Revises: 4a8c2f6e9d13 +Create Date: 2026-10-16 10:27:22.137896+00:00 +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa +import sqlmodel + + +revision: str = '9b3e7d215f60' +down_revision: Union[str, None] = '4a8c2f6e9d13' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table('assistant_session', + sa.Column('id', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('user_id', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('project_id', sqlmodel.sql.sqltypes.AutoString(), nullable=True), + sa.Column('title', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('messages', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('created_at', sa.DateTime(), nullable=False), + sa.Column('updated_at', sa.DateTime(), nullable=False), + sa.ForeignKeyConstraint(['user_id'], ['user.id'], ), + sa.PrimaryKeyConstraint('id') + ) + with op.batch_alter_table('assistant_session', schema=None) as batch_op: + batch_op.create_index(batch_op.f('ix_assistant_session_project_id'), ['project_id'], unique=False) + batch_op.create_index(batch_op.f('ix_assistant_session_user_id'), ['user_id'], unique=False) + + op.create_table('assistant_turn', + sa.Column('id', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('session_id', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('user_id', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('project_id', sqlmodel.sql.sqltypes.AutoString(), nullable=True), + sa.Column('status', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('stage', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('progress', sa.Integer(), nullable=False), + sa.Column('detail', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('error_message', sqlmodel.sql.sqltypes.AutoString(), nullable=True), + sa.Column('prompt', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('base_spec', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('base_version', sa.Integer(), nullable=True), + sa.Column('focus', sqlmodel.sql.sqltypes.AutoString(), nullable=True), + sa.Column('steps', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('reply', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('proposed_spec', sqlmodel.sql.sqltypes.AutoString(), nullable=True), + sa.Column('changes', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('outcome', sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column('cost', sa.Float(), nullable=True), + sa.Column('created_at', sa.DateTime(), nullable=False), + sa.Column('updated_at', sa.DateTime(), nullable=False), + sa.ForeignKeyConstraint(['session_id'], ['assistant_session.id'], ), + sa.ForeignKeyConstraint(['user_id'], ['user.id'], ), + sa.PrimaryKeyConstraint('id') + ) + with op.batch_alter_table('assistant_turn', schema=None) as batch_op: + batch_op.create_index(batch_op.f('ix_assistant_turn_project_id'), ['project_id'], unique=False) + batch_op.create_index(batch_op.f('ix_assistant_turn_session_id'), ['session_id'], unique=False) + batch_op.create_index(batch_op.f('ix_assistant_turn_user_id'), ['user_id'], unique=False) + + + +def downgrade() -> None: + with op.batch_alter_table('assistant_turn', schema=None) as batch_op: + batch_op.drop_index(batch_op.f('ix_assistant_turn_user_id')) + batch_op.drop_index(batch_op.f('ix_assistant_turn_session_id')) + batch_op.drop_index(batch_op.f('ix_assistant_turn_project_id')) + + op.drop_table('assistant_turn') + with op.batch_alter_table('assistant_session', schema=None) as batch_op: + batch_op.drop_index(batch_op.f('ix_assistant_session_user_id')) + batch_op.drop_index(batch_op.f('ix_assistant_session_project_id')) + + op.drop_table('assistant_session') diff --git a/backend/app/assistant/__init__.py b/backend/app/assistant/__init__.py new file mode 100644 index 0000000..3ddaa85 --- /dev/null +++ b/backend/app/assistant/__init__.py @@ -0,0 +1 @@ +"""The video assistant: tools over a working copy of the spec, a loop, a job (M4/M5).""" diff --git a/backend/app/assistant/loop.py b/backend/app/assistant/loop.py new file mode 100644 index 0000000..9b8917a --- /dev/null +++ b/backend/app/assistant/loop.py @@ -0,0 +1,87 @@ +"""One turn: the model and the tools, round after round, until the model is done. + +Pure of the database and the job queue (`worker.py` owns those), so a whole turn can be +run in a test against a scripted provider. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass, field +from typing import Callable, List, Optional, Sequence, Tuple + +from app.assistant import tools as T +from app.assistant.tools import Working, _issues_text +from app.providers.base import ChatProvider, Reply + +logger = logging.getLogger(__name__) + +# Times the loop tells a model that stopped with the video still invalid to fix it. +# Past that it would only be spending the user's money to repeat itself. +MAX_REPAIR_NUDGES = 2 + +Step = Callable[[str, str], None] + + +class Stopped(Exception): + """The user pressed Stop.""" + + +@dataclass +class Outcome: + reply: str + # Set when the turn ran out of rounds or output before finishing on its own. + cut_short: Optional[str] = None + replies: List[Reply] = field(default_factory=list) + + +def _first_line(text: str) -> str: + return text.split("\n", 1)[0].strip() + + +def run(provider: ChatProvider, working: Working, *, system: Sequence[str], + history: Sequence[Tuple[str, str]], prompt: str, on_step: Step, + on_reply: Callable[[Reply], None], cancelled: Callable[[], bool], + max_rounds: int) -> Outcome: + conversation = provider.conversation(system=system, tools=T.SPECS, history=history, + prompt=prompt) + outcome = Outcome(reply="") + nudges = 0 + for _ in range(max_rounds): + if cancelled(): + raise Stopped() + reply = conversation.ask() + outcome.replies.append(reply) + on_reply(reply) + if reply.text: + outcome.reply = reply.text + if not reply.tool_calls: + if reply.stop == "max_tokens": + outcome.cut_short = ("The reply hit the provider's output limit. Raise the " + "maximum response length in Settings, or ask for less " + "at once.") + return outcome + if working.issues and working.changes and nudges < MAX_REPAIR_NUDGES: + nudges += 1 + on_step("issue", f"{len(working.issues)} issue(s) left to fix") + conversation.tell("The video still has issues, so it can't be proposed yet. " + "Fix these, then finish:\n" + _issues_text(working.issues)) + continue + return outcome + if reply.text: + on_step("think", reply.text) + results = [] + for call in reply.tool_calls: + if cancelled(): + raise Stopped() + if call.name == "search_assets": + on_step("search", f"Searching the library: {call.arguments.get('query', '')!r}") + result = T.execute(working, call) + results.append(result) + if call.name in T.READS: + continue + on_step("issue" if result.is_error else "edit", _first_line(result.content)) + conversation.answer(results) + outcome.cut_short = (f"Stopped after {max_rounds} rounds without finishing. What it had " + "done so far is below; ask again to carry on.") + return outcome diff --git a/backend/app/assistant/prompt.py b/backend/app/assistant/prompt.py new file mode 100644 index 0000000..beedf5a --- /dev/null +++ b/backend/app/assistant/prompt.py @@ -0,0 +1,155 @@ +"""What the assistant is told, in two blocks. + +1. `REFERENCE`: who it is and how a VideoSpec works. Fixed text, so a provider that caches + prompts reads it at the cached rate on every turn of every session. +2. `context(...)`: this project as it stands. Fixed for one turn, so the rounds of a turn + share its cache too. + +The reference is written for a capable model: it says what the format means and what +good work looks like, not a checklist of commands. The schema itself is the authority, +and every edit is validated against it, so the reference doesn't try to restate every +field and range. +""" + +from __future__ import annotations + +import json +from typing import Any, Dict, Optional + +REFERENCE = """\ +You are the editor inside Gecko Video Creator (GVC), working with the user on one narrated \ +video. You can build a video from nothing, from a brief, or from material in their asset \ +library (GAM), and you can make any change to an existing one. You work by calling tools \ +that edit a working copy of the video. Nothing you do reaches the project until the user \ +presses Apply on what you propose, so make the edits the request needs, then say what you \ +changed. + +## How a video is described (VideoSpec 0.2) +A video is one JSON document: +- `title`, and project-wide `output` (aspect 16:9 / 9:16 / 1:1, resolution, subtitles, \ +chapters), `voice` (speed, pauses; leave `voice.voice` unset to use the account's voice), \ +`style`, `chrome`, `credits`. +- `assets`: a registry of named keys. Library files are `{"from":"gam","id":,\ +"type":"image"|"video"|"audio","label":...}`. Placeholders are `{"from":"search","query":...}` \ +(footage still to find) or `{"from":"generate","type":"image",...,"prompt":...}`. Segments \ +and tracks refer to assets by key, never by gam_id directly. +- `segments`, in order. Each has a stable `id`, a `label`, and a `kind`: + - `scene`: a `visual` (a still or muted b-roll video) with `narration.beats` read over it. \ +Each beat is one line or paragraph of script: `{"id":"b1","text":"..."}`. A scene lasts as \ +long as its narration, at least `style.min_segment_seconds`. + - `card`: a title or chapter screen: `{"card":{"variant":"title"|"chapter","title":...,\ +"subtitle":...}}`. May have narration too. + - `clip`: a video played with its own sound and no narration: `visual` with `trim` and \ +`"audio":"play"`. + - Optional: `chapter` (a chapter mark at its start), `transition_in`, `duration` \ +(`min` or `fixed` seconds), `overlays`, `role` (title / intro / outro). +- `audio_tracks`: music, sound effects or extra voice, each from a `start` anchor to an \ +`end` anchor (omit `end` to play once to its natural end), with `volume` (0-1; music \ +usually 0.12-0.2 under narration), `fade_in` / `fade_out` seconds, `duck_under_narration`, \ +`loop`. + +## Time is anchored to structure, not seconds +An anchor is `"video_start"`, `"video_end"`, or `{"segment":id,"beat":id?,"edge":"start"|\ +"end","offset":seconds}`. Inside a segment's overlay, a missing `segment` means that segment. \ +Anchoring to structure is what keeps music and overlays on the right cut when scenes are \ +reworded or moved, so always anchor; never compute seconds yourself. +- Change the music partway through: end track 1 at a segment's start and start track 2 \ +there. For a crossfade, start track 2 a few seconds early (negative offset) with a matching \ +`fade_in`, and give track 1 the same `fade_out` with its end a few seconds late. + +## Transitions +`style.transition` is the default join `{style, duration}`; a segment's `transition_in` \ +overrides it field by field. Styles: `none` (a cut), dips that pass through a colour \ +(`fade` to black, `fadewhite`), and blends that overlap the two shots (`dissolve`, \ +`slideleft`/`slideright`, `wipeleft`/`wiperight`, `circleopen`, `smoothleft`). A blend can \ +take at most half of the shorter shot. Use cuts within a sequence, dissolves for a change of \ +place or time, and a fade to black for a change of chapter or mood. + +## Pictures and movement +`visual`: `{"asset":key,"fit":"blur"|"pad"|"crop","motion":...,"trim":{"in":s,"out":s},\ +"loop":true}`. On a still, `motion` is a Ken Burns move: presets `zoom_in`, `zoom_out`, \ +`pan_left`, `pan_right`, `pan_up`, `pan_down` with `amount` (0.02-0.5), or a two-frame path \ +`{"effect":"path","from":{"x":0,"y":0,"w":1},"to":{"x":0.55,"y":0.3,"w":0.35},\ +"ease":"in_out"}` that drifts from one crop of the picture to another (fractions of the \ +picture; the height follows the output's aspect). Use a path to move the eye to the part of \ +a picture the narration is about. `style.motion` `alternate` cycles the presets for stills \ +with no motion of their own. + +## On screen +`overlays` on a segment, each with an `id`, an optional `show {from, to}` (default: the whole \ +segment; `to` may reach into a later segment to hold something across a transition), and \ +`fade` seconds: +- `{"type":"lower_third","title":"Name","subtitle":"Role"}`: for whoever is speaking in a clip. +- `{"type":"text","text":...,"position":"top-left"|...|"center","size_pct":6}`. +- `{"type":"image","asset":key,"position":...,"scale_pct":25,"opacity":1}`. +- `{"type":"quote","text":...,"attribution":...}` and `{"type":"code",...}`: full-screen \ +panels. Anchored to a beat, they cut the scene so the quote gets its own shot. + +## Credits +`credits {"auto":true,"card":true,"description":true,"extra":[...]}` adds an end card and a \ +description crediting each library asset (from its attribution in GAM) plus `extra` lines. \ +Suggest it when the video uses others' material. + +## The library +Use `search_assets` before choosing any picture, footage or music. It searches names, \ +descriptions and what is said in videos: a hit with `said_at` is the moment someone says it, \ +and `set_visual` with that hit's gam_id cuts to that moment for you. Search in a few \ +different phrasings when the first finds little. Prefer real assets; when nothing fits, \ +`add_placeholder` with a clear description of what to find or generate. Never invent a \ +gam_id or an asset key that isn't in the registry. + +## Pacing +Narration reads at about 15 characters a second. A 3-minute video is about 2,500 characters \ +of narration over 10-20 segments. Keep beats to one or two sentences each so pictures can \ +change with the script. Open with a title card; use chapter cards for long videos. + +## Working +- For a new video or a full rewrite, search the library first, then write the whole draft \ +with `replace_spec` in one call, then fix any issues it reports. For anything smaller, use \ +the targeted tools; don't rewrite what you weren't asked to change. +- Keep existing ids. Give new segments ids like `s_market`, beats `b1`, `b2`…, overlays \ +`o_name`, tracks `m_theme`. +- Every edit is checked as you make it. When a result lists issues, fix them before you \ +finish; a video with issues can't be proposed. +- "This scene" means the focused segment, when one is given below. +- If the request is a question rather than a change, answer it without editing. +- Finish with a short reply in plain prose: what you made or changed and anything the user \ +should check (placeholders to fill, pictures to look at). Don't list every tool call; the \ +user sees the changes separately. +""" + + +def _registry_lines(data: Dict[str, Any]) -> str: + lines = [] + for key, ref in (data.get("assets") or {}).items(): + source = ref.get("from") + if source == "gam": + lines.append(f"- {key}: library {ref.get('type') or 'asset'} " + f"{ref.get('label') or ''} (gam_id {ref.get('id')})".rstrip()) + elif source == "search": + lines.append(f"- {key}: placeholder, footage to find: {ref.get('query')!r}") + elif source == "generate": + lines.append(f"- {key}: placeholder, to generate: {ref.get('prompt')!r}") + elif source == "notes": + lines.append(f"- {key}: a file still in Gecko Notes ({ref.get('label') or ref.get('url')})") + return "\n".join(lines) or "(empty)" + + +def context(data: Dict[str, Any], *, focus: Optional[str], from_scratch: bool, + attached: Optional[str] = None) -> str: + """This project, for this turn.""" + parts = [] + if from_scratch: + parts.append("This is a new video: the working copy is a blank starter. Build it " + "from the user's request.") + if focus: + label = next((s.get("label") for s in data.get("segments", []) + if s.get("id") == focus), None) + parts.append(f"Focused segment (\"this scene\"): {focus}" + + (f" ({label})" if label else "")) + if attached: + parts.append("The user attached these library assets to start from:\n" + attached) + parts.append("Asset registry:\n" + _registry_lines(data)) + parts.append("The video now (VideoSpec JSON):\n" + + json.dumps(data, ensure_ascii=False, separators=(",", ":"))) + return "\n\n".join(parts) diff --git a/backend/app/assistant/tools.py b/backend/app/assistant/tools.py new file mode 100644 index 0000000..690e3c9 --- /dev/null +++ b/backend/app/assistant/tools.py @@ -0,0 +1,635 @@ +"""The assistant's operations on a video: the Notes spec doc's §8.2, addressed by id. + +Every edit works on a `Working` copy of the spec, never on the project. Nothing reaches +the project until a person applies the proposal the turn ends with. + +Each edit is checked with `parse_spec` the moment it is made, and the issues go straight +back to the model as the tool's result, so a mistake is fixed in the same turn by the +same model that made it. That is the plan of attack's "one repair round" (D2), +generalised to every edit rather than only a first draft. + +Two rules decide what an edit that produces issues does: +- **A valid spec stays valid.** An edit that would break it is refused, and the spec is + left as it was, so a slip can't wreck a working video. +- **An invalid spec can be repaired step by step.** After `replace_spec` with mistakes + in it, each edit is accepted while issues remain, and the result says which are left. + +Items are addressed by their ids, never by list position. JSON Patch indexes (the Notes +spec doc, open decision 14) point at the wrong item as soon as something is reordered. +""" + +from __future__ import annotations + +import copy +import json +import re +from dataclasses import dataclass, field +from typing import Any, Callable, Dict, List, Optional, Tuple + +from app.gam.client import GamClient, GamError +from app.providers.base import ToolCall, ToolResult, ToolSpec +from app.render.estimate import estimate +from app.spec.validation import SpecIssue, parse_spec + +# A found moment is padded so a cut doesn't clip a word: the editor's `momentTrim`. +MOMENT_PAD = 0.5 +DEFAULT_MOMENT_SECONDS = 5.0 +MAX_SEARCH_HITS = 12 + + +class ToolFailed(Exception): + """The call can't be done as asked; the message goes back to the model.""" + + +def _issues_text(issues: List[SpecIssue], limit: int = 20) -> str: + lines = [f"- {'.'.join(str(p) for p in i.path) or '(spec)'}: {i.message}" + for i in issues[:limit]] + if len(issues) > limit: + lines.append(f"- … and {len(issues) - limit} more") + return "\n".join(lines) + + +def _slug(text: str) -> str: + return re.sub(r"[^a-z0-9]+", "_", (text or "").lower()).strip("_")[:32] + + +def merge(base: Any, changes: Any) -> Any: + """A JSON merge patch (RFC 7386): objects merge, a null removes, anything else + replaces. Lists are replaced whole, which is what "these are the beats" means.""" + if not isinstance(changes, dict) or not isinstance(base, dict): + return copy.deepcopy(changes) + out = copy.deepcopy(base) + for key, value in changes.items(): + if value is None: + out.pop(key, None) + else: + out[key] = merge(out.get(key), value) + return out + + +@dataclass +class Hit: + """What a search found, kept so `set_visual` can take its moment as the trim.""" + + gam_id: str + name: str + asset_type: str + duration: Optional[float] + start: Optional[float] + end: Optional[float] + + +@dataclass +class Working: + data: Dict[str, Any] + gam: Optional[GamClient] = None + changes: List[str] = field(default_factory=list) + hits: Dict[str, Hit] = field(default_factory=dict) + issues: List[SpecIssue] = field(default_factory=list) + + def __post_init__(self) -> None: + self.data = copy.deepcopy(self.data) + _, self.issues = parse_spec(self.data) + + @property + def valid(self) -> bool: + return not self.issues + + # ── lookups ────────────────────────────────────────────────────────────── + + def segments(self) -> List[Dict[str, Any]]: + return self.data.setdefault("segments", []) + + def segment(self, segment_id: str) -> Dict[str, Any]: + for segment in self.segments(): + if segment.get("id") == segment_id: + return segment + known = ", ".join(s.get("id", "?") for s in self.segments()) or "none" + raise ToolFailed(f"No segment {segment_id!r}. The segments are: {known}.") + + def index_of(self, segment_id: str) -> int: + return self.segments().index(self.segment(segment_id)) + + def name_of(self, segment: Dict[str, Any]) -> str: + return segment.get("label") or segment.get("id", "?") + + def tracks(self) -> List[Dict[str, Any]]: + return self.data.setdefault("audio_tracks", []) + + def track(self, track_id: str) -> Dict[str, Any]: + for track in self.tracks(): + if track.get("id") == track_id: + return track + raise ToolFailed(f"No audio track {track_id!r}.") + + def new_id(self, prefix: str, taken: List[str]) -> str: + n = len(taken) + 1 + while f"{prefix}{n}" in taken: + n += 1 + return f"{prefix}{n}" + + def all_overlay_ids(self) -> List[str]: + return [o.get("id") for s in self.segments() for o in s.get("overlays", [])] + + def register_gam(self, data: Dict[str, Any], gam_id: str, *, + asset_type: Optional[str] = None, label: Optional[str] = None) -> str: + """The asset key for a GAM asset in `data`, added to its registry if new. + + `data` rather than `self.data`: edits build a candidate and commit it whole, so + a refused edit leaves no stray registry entry behind. + """ + assets = data.setdefault("assets", {}) + for key, ref in assets.items(): + if ref.get("from") == "gam" and ref.get("id") == gam_id: + return key + hit = self.hits.get(gam_id) + label = label or (hit.name if hit else None) + asset_type = asset_type or (hit.asset_type if hit else None) + base = f"a_{_slug(label or '') or 'gam'}" + key, n = base, 2 + while key in assets: + key, n = f"{base}_{n}", n + 1 + ref: Dict[str, Any] = {"from": "gam", "id": gam_id} + if asset_type in ("image", "video", "audio"): + ref["type"] = asset_type + if label: + ref["label"] = label + assets[key] = ref + return key + + # ── committing ─────────────────────────────────────────────────────────── + + def commit(self, candidate: Dict[str, Any], summary: str) -> str: + """Take `candidate` as the new spec if the rules above allow it.""" + _, issues = parse_spec(candidate) + if issues and self.valid: + raise ToolFailed("That edit would make the video invalid, so it was not made:\n" + + _issues_text(issues)) + self.data, self.issues = candidate, issues + self.changes.append(summary) + if issues: + return (f"Done: {summary}. The video still has issues to fix:\n" + + _issues_text(issues)) + return f"Done: {summary}." + + +# ── the tools ───────────────────────────────────────────────────────────────── + +def _compact(data: Dict[str, Any]) -> str: + return json.dumps(data, ensure_ascii=False, separators=(",", ":")) + + +def get_spec(w: Working, segment_id: Optional[str] = None) -> str: + if segment_id: + return _compact(w.segment(segment_id)) + return _compact(w.data) + + +def search_assets(w: Working, query: str, type: Optional[str] = None, + limit: int = 8) -> str: + if w.gam is None: + raise ToolFailed("The asset library isn't reachable in this turn.") + try: + found = w.gam.search(query, asset_type=type, limit=max(1, min(MAX_SEARCH_HITS, limit))) + except GamError as exc: + raise ToolFailed(f"The asset library search failed: {exc.message}") + rows = [] + for hit in found.data: + asset = hit.asset + w.hits[asset.id] = Hit(asset.id, asset.name, asset.asset_type, asset.duration_seconds, + hit.start_time, hit.end_time) + row: Dict[str, Any] = {"gam_id": asset.id, "name": asset.name, "type": asset.asset_type} + if asset.duration_seconds: + row["duration"] = round(asset.duration_seconds, 1) + if hit.start_time is not None: + # Where in the video it was said: what `set_visual` will cut to. + row["said_at"] = [round(hit.start_time, 1), + round(hit.end_time, 1) if hit.end_time is not None else None] + if hit.snippet: + row["snippet"] = hit.snippet.replace("«", "").replace("»", "")[:240] + if asset.credit: + row["credit"] = asset.credit + if asset.width and asset.height: + row["size"] = f"{asset.width}x{asset.height}" + rows.append(row) + if not rows: + return (f"Nothing in the library matches {query!r}. Try other words, or leave a " + "placeholder with add_placeholder.") + return _compact({"query": query, "semantic": found.semantic, "hits": rows}) + + +def get_asset(w: Working, gam_id: str) -> str: + if w.gam is None: + raise ToolFailed("The asset library isn't reachable in this turn.") + try: + asset = w.gam.get_asset(gam_id) + except GamError as exc: + raise ToolFailed(f"The asset library failed: {exc.message}") + if asset is None: + raise ToolFailed(f"No asset {gam_id!r} in the library.") + return _compact({"gam_id": asset.id, "name": asset.name, "type": asset.asset_type, + "duration": asset.duration_seconds, "width": asset.width, + "height": asset.height, "credit": asset.credit, + "clip_of": asset.parent_asset_id}) + + +def replace_spec(w: Working, spec: Dict[str, Any]) -> str: + if not isinstance(spec, dict): + raise ToolFailed("spec must be a VideoSpec object.") + candidate = copy.deepcopy(spec) + # The project owns its id; a draft never changes it. + if "id" in w.data: + candidate["id"] = w.data["id"] + # Accepted even with issues: a whole draft is the one edit too big to throw away + # over a typo. The issues come back to be fixed with the other tools. + _, issues = parse_spec(candidate) + w.data, w.issues = candidate, issues + w.changes.append(f"Drafted the video: {len(candidate.get('segments', []))} segments") + if issues: + return (f"The draft is in place, with {len(issues)} issue" + f"{'s' if len(issues) != 1 else ''} to fix:\n" + _issues_text(issues)) + return "Done: the draft is in place and valid." + + +def _insert_at(w: Working, after: Optional[str]) -> int: + if after in (None, ""): + return len(w.segments()) + if after == "start": + return 0 + return w.index_of(after) + 1 + + +def add_segment(w: Working, segment: Dict[str, Any], after: Optional[str] = None) -> str: + if not isinstance(segment, dict): + raise ToolFailed("segment must be an object.") + candidate = copy.deepcopy(w.data) + taken = [s.get("id") for s in candidate.get("segments", [])] + segment = copy.deepcopy(segment) + if not segment.get("id") or segment["id"] in taken: + segment["id"] = w.new_id("s_", taken) + position = _insert_at(w, after) + candidate.setdefault("segments", []).insert(position, segment) + return w.commit(candidate, f"Added {segment.get('kind', 'segment')} " + f"{segment.get('label') or segment['id']} (id {segment['id']})") + + +def update_segment(w: Working, segment_id: str, changes: Dict[str, Any]) -> str: + if not isinstance(changes, dict): + raise ToolFailed("changes must be an object.") + if "id" in changes: + raise ToolFailed("A segment's id can't be changed; anchors point at it.") + candidate = copy.deepcopy(w.data) + index = w.index_of(segment_id) + candidate["segments"][index] = merge(candidate["segments"][index], changes) + fields = ", ".join(sorted(changes)) or "nothing" + return w.commit(candidate, f"Changed {w.name_of(candidate['segments'][index])}: {fields}") + + +def move_segment(w: Working, segment_id: str, after: str) -> str: + candidate = copy.deepcopy(w.data) + index = w.index_of(segment_id) + moving = candidate["segments"].pop(index) + if after == segment_id: + raise ToolFailed("A segment can't be moved after itself.") + if after == "start": + position = 0 + else: + ids = [s.get("id") for s in candidate["segments"]] + if after not in ids: + raise ToolFailed(f"No segment {after!r} to move it after.") + position = ids.index(after) + 1 + candidate["segments"].insert(position, moving) + where = "to the start" if after == "start" else f"after {after}" + return w.commit(candidate, f"Moved {w.name_of(moving)} {where}") + + +def _anchors_naming(data: Dict[str, Any], segment_id: str) -> List[str]: + """What points at a segment, so removing it can say what will be orphaned.""" + named = [] + + def hits(anchor: Any) -> bool: + return isinstance(anchor, dict) and anchor.get("segment") == segment_id + + for track in data.get("audio_tracks", []): + if hits(track.get("start")) or hits(track.get("end")): + named.append(f"audio track {track.get('id')}") + for segment in data.get("segments", []): + for overlay in segment.get("overlays", []): + show = overlay.get("show") or {} + if hits(show.get("from")) or hits(show.get("to")): + named.append(f"overlay {overlay.get('id')} on {segment.get('id')}") + return named + + +def remove_segment(w: Working, segment_id: str) -> str: + candidate = copy.deepcopy(w.data) + index = w.index_of(segment_id) + removed = candidate["segments"].pop(index) + result = w.commit(candidate, f"Removed {w.name_of(removed)}") + orphans = _anchors_naming(candidate, segment_id) + if orphans: + # Allowed, as in the editor: the renderer drops what can't be placed with a + # warning. Said so the model can re-anchor them. + result += (" These are still anchored to it and will be left out of the render " + "unless re-anchored: " + ", ".join(orphans) + ".") + return result + + +def _moment_trim(hit: Hit) -> Optional[Dict[str, float]]: + if hit.start is None: + return None + end = hit.end if hit.end is not None else hit.start + DEFAULT_MOMENT_SECONDS + a = max(0.0, hit.start - MOMENT_PAD) + b = end + MOMENT_PAD + if hit.duration: + b = min(b, hit.duration) + return {"in": round(a, 1), "out": round(b, 1)} if b > a else None + + +def set_visual(w: Working, segment_id: str, gam_id: Optional[str] = None, + asset: Optional[str] = None, trim: Optional[Dict[str, Any]] = None, + motion: Optional[Dict[str, Any]] = None, fit: Optional[str] = None, + audio: Optional[str] = None, loop: Optional[bool] = None) -> str: + if not gam_id and not asset: + raise ToolFailed("Give gam_id (from search_assets) or asset (a key in the registry).") + candidate = copy.deepcopy(w.data) + if gam_id: + key = w.register_gam(candidate, gam_id) + else: + if asset not in w.data.get("assets", {}): + raise ToolFailed(f"No asset {asset!r} in the registry.") + key = asset + index = w.index_of(segment_id) + segment = candidate["segments"][index] + visual: Dict[str, Any] = {"asset": key} + previous = segment.get("visual") or {} + if previous.get("asset") == key: + visual = copy.deepcopy(previous) + hit = w.hits.get(gam_id) if gam_id else None + if trim is not None: + visual["trim"] = trim + elif hit and hit.asset_type == "video" and _moment_trim(hit): + # The moment that was said becomes the cut, as picking a hit does in the editor. + visual["trim"] = _moment_trim(hit) + if motion is not None: + visual["motion"] = motion + if fit is not None: + visual["fit"] = fit + if audio is not None: + visual["audio"] = audio + if loop is not None: + visual["loop"] = loop + segment["visual"] = visual + label = hit.name if hit else key + trimmed = f", {visual['trim'].get('in')}–{visual['trim'].get('out')} s" if visual.get("trim") else "" + return w.commit(candidate, f"{w.name_of(segment)} shows {label}{trimmed}") + + +def add_placeholder(w: Working, segment_id: str, kind: str, text: str, + media_type: str = "video", moment_min: Optional[float] = None, + moment_max: Optional[float] = None, use_as: str = "visual") -> str: + """Footage still to find, or a picture still to generate: drawn as a slate in a + preview, filled in by a person later.""" + candidate = copy.deepcopy(w.data) + assets = candidate.setdefault("assets", {}) + base = f"a_{_slug(text) or kind}" + key, n = base, 2 + while key in assets: + key, n = f"{base}_{n}", n + 1 + if kind == "search": + ref: Dict[str, Any] = {"from": "search", "query": text, "type": media_type} + if media_type == "video" and (moment_min or moment_max): + ref["moment"] = {k: v for k, v in (("min", moment_min), ("max", moment_max)) if v} + elif kind == "generate": + if media_type not in ("image", "video", "music", "sfx"): + media_type = "image" + ref = {"from": "generate", "type": media_type, "prompt": text} + else: + raise ToolFailed('kind must be "search" or "generate".') + assets[key] = ref + index = w.index_of(segment_id) + segment = candidate["segments"][index] + if use_as == "visual": + segment["visual"] = {"asset": key} + return w.commit(candidate, f"{w.name_of(segment)}: placeholder to " + f"{'find' if kind == 'search' else 'generate'}: {text!r}") + + +def add_overlay(w: Working, segment_id: str, overlay: Dict[str, Any], + gam_id: Optional[str] = None) -> str: + if not isinstance(overlay, dict): + raise ToolFailed("overlay must be an object.") + candidate = copy.deepcopy(w.data) + overlay = copy.deepcopy(overlay) + if gam_id: + overlay["asset"] = w.register_gam(candidate, gam_id, asset_type="image") + taken = w.all_overlay_ids() + if not overlay.get("id") or overlay["id"] in taken: + overlay["id"] = w.new_id("o_", taken) + index = w.index_of(segment_id) + segment = candidate["segments"][index] + segment.setdefault("overlays", []).append(overlay) + what = overlay.get("title") or overlay.get("text") or overlay.get("type") + return w.commit(candidate, f"On {w.name_of(segment)}: added " + f"{overlay.get('type', 'overlay')} {what!r} (id {overlay['id']})") + + +def _overlay_index(segment: Dict[str, Any], overlay_id: str) -> int: + for i, overlay in enumerate(segment.get("overlays", [])): + if overlay.get("id") == overlay_id: + return i + raise ToolFailed(f"No overlay {overlay_id!r} on {segment.get('id')}.") + + +def update_overlay(w: Working, segment_id: str, overlay_id: str, changes: Dict[str, Any]) -> str: + candidate = copy.deepcopy(w.data) + segment = candidate["segments"][w.index_of(segment_id)] + i = _overlay_index(segment, overlay_id) + segment["overlays"][i] = merge(segment["overlays"][i], changes) + return w.commit(candidate, f"On {w.name_of(segment)}: changed overlay {overlay_id} " + f"({', '.join(sorted(changes))})") + + +def remove_overlay(w: Working, segment_id: str, overlay_id: str) -> str: + candidate = copy.deepcopy(w.data) + segment = candidate["segments"][w.index_of(segment_id)] + segment["overlays"].pop(_overlay_index(segment, overlay_id)) + return w.commit(candidate, f"On {w.name_of(segment)}: removed overlay {overlay_id}") + + +def add_audio_track(w: Working, track: Dict[str, Any], gam_id: Optional[str] = None) -> str: + if not isinstance(track, dict): + raise ToolFailed("track must be an object.") + candidate = copy.deepcopy(w.data) + track = copy.deepcopy(track) + if gam_id: + track["asset"] = w.register_gam(candidate, gam_id, asset_type="audio") + taken = [t.get("id") for t in candidate.get("audio_tracks", [])] + if not track.get("id") or track["id"] in taken: + track["id"] = w.new_id("m_" if track.get("kind", "music") == "music" else "x_", taken) + candidate.setdefault("audio_tracks", []).append(track) + hit = w.hits.get(gam_id) if gam_id else None + return w.commit(candidate, f"Added {track.get('kind', 'music')} track " + f"{(hit.name if hit else track.get('asset'))!r} (id {track['id']})") + + +def update_audio_track(w: Working, track_id: str, changes: Dict[str, Any]) -> str: + candidate = copy.deepcopy(w.data) + tracks = candidate.get("audio_tracks", []) + for i, track in enumerate(tracks): + if track.get("id") == track_id: + tracks[i] = merge(track, changes) + return w.commit(candidate, f"Changed audio track {track_id} " + f"({', '.join(sorted(changes))})") + raise ToolFailed(f"No audio track {track_id!r}.") + + +def remove_audio_track(w: Working, track_id: str) -> str: + w.track(track_id) + candidate = copy.deepcopy(w.data) + candidate["audio_tracks"] = [t for t in candidate["audio_tracks"] if t.get("id") != track_id] + return w.commit(candidate, f"Removed audio track {track_id}") + + +_PROJECT_KEYS = ("title", "style", "output", "voice", "chrome", "credits") + + +def set_project(w: Working, changes: Dict[str, Any]) -> str: + if not isinstance(changes, dict) or not changes: + raise ToolFailed("changes must be an object.") + unknown = sorted(set(changes) - set(_PROJECT_KEYS)) + if unknown: + raise ToolFailed(f"set_project changes only {', '.join(_PROJECT_KEYS)}; " + f"not {', '.join(unknown)}. Segments and tracks have their own tools.") + candidate = merge(w.data, changes) + return w.commit(candidate, f"Changed the project's {', '.join(sorted(changes))}") + + +def run_estimate(w: Working) -> str: + spec, issues = parse_spec(w.data) + if spec is None: + return "Can't estimate yet; the video has issues:\n" + _issues_text(issues) + result = estimate(spec) + return _compact({"seconds": result.seconds, "seconds_is_floor": result.seconds_is_floor, + "narration_chars": result.narration_chars, + "blocks_full_render": result.blocks_render}) + + +# ── schemas ─────────────────────────────────────────────────────────────────── + +_OBJ = {"type": "object"} +_STR = {"type": "string"} +_NUM = {"type": "number"} + +TOOLS: List[Tuple[ToolSpec, Callable[..., str]]] = [ + (ToolSpec("get_spec", "The video as it stands now (after this turn's edits), or one " + "segment of it.", + {"type": "object", "properties": {"segment_id": _STR}}), get_spec), + (ToolSpec("search_assets", + "Search the user's asset library (GAM) by name, description and what is SAID in " + "videos. A hit on speech carries said_at [start, end] seconds: the moment to cut " + "to. Use type to restrict to video, image or audio.", + {"type": "object", "properties": { + "query": _STR, "type": {"type": "string", "enum": ["video", "image", "audio"]}, + "limit": {"type": "integer", "minimum": 1, "maximum": MAX_SEARCH_HITS}}, + "required": ["query"]}), search_assets), + (ToolSpec("get_asset", "Details of one library asset by gam_id.", + {"type": "object", "properties": {"gam_id": _STR}, "required": ["gam_id"]}), + get_asset), + (ToolSpec("replace_spec", + "Replace the whole video with a complete VideoSpec. For a first draft or a full " + "rewrite only; use the targeted tools for anything smaller. Library assets go in " + "`assets` as {\"from\":\"gam\",\"id\":,\"type\":...}.", + {"type": "object", "properties": {"spec": _OBJ}, "required": ["spec"]}), + replace_spec), + (ToolSpec("add_segment", + "Insert a segment (scene, card or clip) after the segment `after` (\"start\" for " + "first, omit for last). A missing or taken id is replaced with a fresh one.", + {"type": "object", "properties": {"segment": _OBJ, "after": _STR}, + "required": ["segment"]}), add_segment), + (ToolSpec("update_segment", + "Change fields of one segment by merge patch: objects merge, null removes a " + "field, lists (beats, overlays) are replaced whole.", + {"type": "object", "properties": {"segment_id": _STR, "changes": _OBJ}, + "required": ["segment_id", "changes"]}), update_segment), + (ToolSpec("move_segment", "Move a segment to after another (\"start\" for first).", + {"type": "object", "properties": {"segment_id": _STR, "after": _STR}, + "required": ["segment_id", "after"]}), move_segment), + (ToolSpec("remove_segment", "Remove a segment. Says what was anchored to it.", + {"type": "object", "properties": {"segment_id": _STR}, + "required": ["segment_id"]}), remove_segment), + (ToolSpec("set_visual", + "Set a segment's picture or footage: gam_id (from search_assets, registered for " + "you) or asset (an existing registry key). A video hit's said_at becomes the trim " + "unless trim is given. motion and fit as in the spec.", + {"type": "object", "properties": { + "segment_id": _STR, "gam_id": _STR, "asset": _STR, "trim": _OBJ, + "motion": _OBJ, "fit": {"type": "string", "enum": ["blur", "pad", "crop"]}, + "audio": {"type": "string", "enum": ["mute", "play"]}, + "loop": {"type": "boolean"}}, + "required": ["segment_id"]}), set_visual), + (ToolSpec("add_placeholder", + "When the library has nothing fitting: a placeholder for footage still to find " + "(kind search, text = what to look for) or a picture to generate (kind generate, " + "text = the prompt), set as the segment's visual. Never invent a gam_id.", + {"type": "object", "properties": { + "segment_id": _STR, "kind": {"type": "string", "enum": ["search", "generate"]}, + "text": _STR, "media_type": {"type": "string", + "enum": ["video", "image", "audio", "music", "sfx"]}, + "moment_min": _NUM, "moment_max": _NUM}, + "required": ["segment_id", "kind", "text"]}), add_placeholder), + (ToolSpec("add_overlay", + "Add an overlay to a segment: text, lower_third, image (pass gam_id for the " + "picture), quote or code. show {from, to} anchors default to the segment.", + {"type": "object", "properties": {"segment_id": _STR, "overlay": _OBJ, + "gam_id": _STR}, + "required": ["segment_id", "overlay"]}), add_overlay), + (ToolSpec("update_overlay", "Change an overlay by merge patch.", + {"type": "object", "properties": {"segment_id": _STR, "overlay_id": _STR, + "changes": _OBJ}, + "required": ["segment_id", "overlay_id", "changes"]}), update_overlay), + (ToolSpec("remove_overlay", "Remove an overlay.", + {"type": "object", "properties": {"segment_id": _STR, "overlay_id": _STR}, + "required": ["segment_id", "overlay_id"]}), remove_overlay), + (ToolSpec("add_audio_track", + "Add music, a sound effect or voice from the library (gam_id from search_assets " + "with type audio), with start/end anchors, volume and fades.", + {"type": "object", "properties": {"track": _OBJ, "gam_id": _STR}, + "required": ["track"]}), add_audio_track), + (ToolSpec("update_audio_track", "Change an audio track by merge patch.", + {"type": "object", "properties": {"track_id": _STR, "changes": _OBJ}, + "required": ["track_id", "changes"]}), update_audio_track), + (ToolSpec("remove_audio_track", "Remove an audio track.", + {"type": "object", "properties": {"track_id": _STR}, "required": ["track_id"]}), + remove_audio_track), + (ToolSpec("set_project", + "Change project-wide settings by merge patch: title, style (default transition, " + "motion, background, cards…), output, voice, chrome, credits.", + {"type": "object", "properties": {"changes": _OBJ}, "required": ["changes"]}), + set_project), + (ToolSpec("estimate", "How long the video runs, its narration length, and what would " + "stop a full render.", {"type": "object", "properties": {}}), run_estimate), +] + +SPECS = [spec for spec, _ in TOOLS] +_HANDLERS = {spec.name: handler for spec, handler in TOOLS} + +# Read-only tools, which neither change the video nor count as an edit in the steps. +READS = {"get_spec", "search_assets", "get_asset", "estimate"} + + +def execute(w: Working, call: ToolCall) -> ToolResult: + """Run one call. Never raises: a failure is a result the model reads and acts on.""" + if call.parse_error: + return ToolResult(call.id, f"Your arguments couldn't be read: {call.parse_error}.", + is_error=True) + handler = _HANDLERS.get(call.name) + if handler is None: + return ToolResult(call.id, f"There is no tool {call.name!r}.", is_error=True) + try: + return ToolResult(call.id, handler(w, **call.arguments)) + except ToolFailed as exc: + return ToolResult(call.id, str(exc), is_error=True) + except TypeError as exc: + # Arguments the handler doesn't take: said in the schema's terms. + return ToolResult(call.id, f"Wrong arguments for {call.name}: {exc}", is_error=True) diff --git a/backend/app/assistant/worker.py b/backend/app/assistant/worker.py new file mode 100644 index 0000000..6d610f5 --- /dev/null +++ b/backend/app/assistant/worker.py @@ -0,0 +1,194 @@ +"""Running assistant turns in the background, as the user who asked. + +A turn searches GAM with the requester's own token, so it can never see what they could +not. The token is held here, in memory, for the turn's life only, never in the database, +which is why a server restart ends a running turn ("ask again") rather than resuming it: +the render queue's rule, for the same reason. + +Everything else lives on the `AssistantTurn` row as it happens, so a turn finishes and +waits for Apply with nobody watching. +""" + +from __future__ import annotations + +import json +import logging +import uuid +from typing import Dict, List, Optional, Tuple + +from sqlmodel import Session + +from app.assistant import loop +from app.assistant.prompt import REFERENCE, context +from app.assistant.tools import Working +from app.clock import utcnow +from app.config import settings +from app.gam.client import GamClient +from app.jobs.runner import JobQueue, readable_error, set_fields +from app.models.assistant import AssistantSession, AssistantTurn +from app.providers import build_provider +from app.providers.base import ProviderError, Reply +from app.spec.model import dump_spec +from app.spec.validation import parse_spec +from app.usage import events as usage_events + +logger = logging.getLogger(__name__) + +_tokens: Dict[str, Optional[str]] = {} + +MAX_STEPS_KEPT = 200 + + +def append_message(row: AssistantSession, role: str, text: str, + turn_id: Optional[str] = None) -> None: + messages = json.loads(row.messages or "[]") + messages.append({"role": role, "text": text, "turn_id": turn_id, + "at": utcnow().isoformat(timespec="seconds")}) + row.messages = json.dumps(messages, ensure_ascii=False) + row.updated_at = utcnow() + + +def history_for(row: AssistantSession, turn_id: str) -> List[Tuple[str, str]]: + """Earlier messages, newest kept, with this turn's own prompt left out (it is sent + as the new message).""" + messages = [m for m in json.loads(row.messages or "[]") + if m.get("turn_id") != turn_id and m.get("text") + and m.get("role") in ("user", "assistant")] + return [(m["role"], m["text"]) for m in messages[-settings.assistant_history_messages:]] + + +def _canonical(data: dict) -> Optional[dict]: + spec, issues = parse_spec(data) + return dump_spec(spec) if spec is not None else None + + +def _run(turn_id: str) -> None: + with queue.session() as session: + turn = session.get(AssistantTurn, turn_id) + if turn is None or turn.status != "queued": + return + chat = session.get(AssistantSession, turn.session_id) + if turn_id not in _tokens: + set_fields(session, turn, status="error", stage="", + error_message="Interrupted by a restart of the server. Ask again.") + if chat: + append_message(chat, "assistant", "_Interrupted by a restart. Ask again._", turn_id) + session.add(chat) + session.commit() + return + set_fields(session, turn, status="processing", stage="Thinking", progress=0) + user_id, project_id = turn.user_id, turn.project_id + prompt, focus = turn.prompt, turn.focus + base = json.loads(turn.base_spec or "{}") + history = history_for(chat, turn_id) if chat else [] + + token = _tokens.get(turn_id) + gam = GamClient(token) + steps: List[dict] = [] + cost: List[float] = [] + + def write(**fields) -> None: + with queue.session() as session: + row = session.get(AssistantTurn, turn_id) + if row is not None: + set_fields(session, row, **fields) + + def on_step(kind: str, text: str) -> None: + steps.append({"kind": kind, "text": text[:600]}) + detail = {"search": "searching the library", "edit": "editing", + "issue": "fixing", "think": "thinking"}.get(kind, "") + write(steps=json.dumps(steps[-MAX_STEPS_KEPT:], ensure_ascii=False), + stage="Working", detail=detail) + + def on_reply(reply: Reply) -> None: + with queue.session() as session: + spent = usage_events.record_reply(session, user_id=user_id, project_id=project_id, + turn_id=turn_id, reply=reply) + if spent is not None: + cost.append(spent) + write(cost=round(sum(cost), 6)) + + try: + with queue.session() as session: + provider = build_provider(session, user_id) + if provider is None: + raise ProviderError("No AI provider is set up. Add one in Settings first.") + working = Working(base, gam=gam) + system = [REFERENCE, context(base, focus=focus, from_scratch=project_id is None)] + outcome = loop.run(provider, working, system=system, history=history, prompt=prompt, + on_step=on_step, on_reply=on_reply, + cancelled=lambda: queue.is_cancelled(turn_id), + max_rounds=settings.assistant_max_rounds) + except loop.Stopped: + _finish(turn_id, status="cancelled", stage="", message="_Stopped._", steps=steps) + return + except ProviderError as exc: + _finish(turn_id, status="error", stage="", error_message=str(exc), + message=f"_{exc}_", steps=steps) + return + except Exception as exc: # noqa: BLE001 - a failed turn is a job error, not a crash + logger.exception("Assistant turn %s failed", turn_id) + _finish(turn_id, status="error", stage="", error_message=readable_error(exc), + message="_Something went wrong; ask again._", steps=steps) + return + finally: + gam.close() + _tokens.pop(turn_id, None) + + reply = outcome.reply or ("Done." if working.changes else "") + proposed = None + if working.changes: + canonical = _canonical(working.data) + if canonical is None: + reply += ("\n\nI couldn't get the video into a valid state, so there's nothing to " + "apply. Ask me to try again, or say what to change.") + elif canonical != _canonical(base): + proposed = canonical + if outcome.cut_short: + reply = (reply + "\n\n" if reply else "") + f"_{outcome.cut_short}_" + _finish(turn_id, status="done", stage="", progress=100, detail="", message=reply, + steps=steps, reply=reply, + proposed_spec=json.dumps(proposed, ensure_ascii=False) if proposed else None, + changes=json.dumps(working.changes if proposed else [], ensure_ascii=False), + outcome="pending" if proposed else "") + + +def _finish(turn_id: str, *, message: str, steps: List[dict], **fields) -> None: + with queue.session() as session: + turn = session.get(AssistantTurn, turn_id) + if turn is None: + return + set_fields(session, turn, steps=json.dumps(steps[-MAX_STEPS_KEPT:], ensure_ascii=False), + **fields) + chat = session.get(AssistantSession, turn.session_id) + if chat is not None and message: + append_message(chat, "assistant", message, turn_id) + session.add(chat) + session.commit() + + +queue = JobQueue(AssistantTurn, _run, name="assistant", + concurrency=settings.assistant_concurrency, requeue_on_restart=False) + + +def submit(session: Session, chat: AssistantSession, *, user_id: str, token: Optional[str], + prompt: str, base_spec: dict, base_version: Optional[int], + focus: Optional[str]) -> AssistantTurn: + turn = AssistantTurn(id=str(uuid.uuid4()), session_id=chat.id, user_id=user_id, + project_id=chat.project_id, prompt=prompt, + base_spec=json.dumps(base_spec, ensure_ascii=False), + base_version=base_version, focus=focus) + append_message(chat, "user", prompt, turn.id) + if not chat.title: + chat.title = prompt.strip().splitlines()[0][:80] + session.add(turn) + session.add(chat) + session.commit() + session.refresh(turn) + _tokens[turn.id] = token + queue.enqueue(turn.id) + return turn + + +def start() -> None: + queue.start() diff --git a/backend/app/config.py b/backend/app/config.py index 4bf4398..6ad5c64 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -86,6 +86,15 @@ class Settings(BaseSettings): render_cache_max_gb: float = 20 # The largest single file a render will fetch from GAM. render_max_file_mb: int = 8192 + # The assistant (M4/M5). A turn is a job: it keeps running with the browser closed. + # Rounds are model calls in one turn; a from-scratch draft takes a handful, the + # rest of the budget is for searching and fixing. Two workers: a turn is + # mostly waiting on the provider, so two run side by side without contention. + assistant_concurrency: int = 2 + assistant_max_rounds: int = 24 + # Earlier messages a turn is given, newest kept. Older ones are still in the + # session; the spec itself, not the chat, is the record of what was decided. + assistant_history_messages: int = 20 job_heartbeat_seconds: int = 30 job_stale_minutes: int = 40 diff --git a/backend/app/main.py b/backend/app/main.py index 6d957b4..b0001b3 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -20,10 +20,13 @@ from app.config import settings from app.database import engine, get_session from app.gam.client import GamError +from app.routers import assistant as assistant_router from app.routers import gam as gam_router from app.routers import projects as projects_router +from app.routers import providers as providers_router from app.routers import renders as renders_router from app.routers import spec as spec_router +from app.routers import usage as usage_router from app.schemas import ClientConfig, DataResponse, HealthResponse logging.basicConfig(level=logging.INFO) @@ -48,9 +51,13 @@ async def lifespan(_app: FastAPI): # Picks up nothing a restart interrupted (a render runs as its requester, whose # token is gone), and starts the render workers. + from app.assistant import worker as assistant_worker from app.render import worker worker.start() + # The same rule for the assistant: a turn interrupted by a restart is ended and + # said so, never rerun without the token it needs. + assistant_worker.start() yield @@ -145,5 +152,8 @@ def me(user: CurrentUser, session: Session = Depends(get_session)) -> DataRespon app.include_router(spec_router.router, prefix="/api/spec", tags=["spec"]) app.include_router(gam_router.router, prefix="/api/gam", tags=["gam"]) app.include_router(renders_router.router, prefix="/api", tags=["renders"]) +app.include_router(providers_router.router, prefix="/api/providers", tags=["providers"]) +app.include_router(assistant_router.router, prefix="/api/assistant", tags=["assistant"]) +app.include_router(usage_router.router, prefix="/api/usage", tags=["usage"]) # Not under /api: these URLs go straight into