Source code for falaw.bridges.skill

"""Render a Claude SKILL.md (and references) from the falaw tool registry.

The skill is the entry point that teaches Claude Code how to use falaw. It
is generated --- never hand-edited at the install location --- so adding a
new tool automatically updates the skill the next time we run
`write_skill_files()`.
"""

from __future__ import annotations

import os
from typing import Iterable, Optional

from ..base import ToolSpec
from ..registry import list_models, list_tools

_HEADER = """---
name: falaw
description: >-
  Generate and manage AI media (images, video, audio) via fal.ai. Use this
  skill whenever the user wants to generate, edit, upscale, or compose media,
  or asks about fal.ai or fal-client. Also use it for *directorial* workflows:
  the user defines a Scene as data (characters, beats, shots) and falaw renders
  it with caching, so a single edit re-renders only what changed.
---

# falaw

Two ways to use `falaw`:

1. **Single-shot operations** --- `generate_image`, `text_to_speech`, etc. for
   one-off media generation.
2. **Directorial workflow** --- author a Scene as data, render it, give notes
   that re-edit the Scene, re-render only the affected beats. This is the path
   to *direct* a film instead of just *generate* clips.

## Directorial workflow (when the user wants to "plan, parametrize, render")

The thesis: keep the film in *editable structure* all the way down to the
pixels. The creator authors a Scene as data; a directorial note becomes a
single IR edit; the renderer caches everything content-addressed so unchanged
beats don't re-render.

### Phase 1: Plan (LLM-assisted or hand-built)

```python
from falaw import parse_screenplay, Scene, Character, Environment, make_beat, make_shot

# Option A: feed prose / treatment text and let any-llm draft the structure.
scene = parse_screenplay(
    prose_text, title="Diner Encounter", style="Wes-Anderson pastel"
)

# Option B: build the Scene directly.
scene = Scene(
    title="Diner Encounter",
    style="Wes-Anderson symmetrical pastel",
    characters=(Character(name="Sarah", description="mid-30s, dark curly hair"),),
    environments=(
        Environment(
            name="diner", description="1950s chrome diner", time_of_day="midnight"
        ),
    ),
    shots=(
        make_shot(
            "two-shot at the booth",
            framing="medium",
            environment="diner",
            characters=("Sarah", "Tom"),
            index=0,
        ),
    ),
    beats=(
        make_beat("Sarah", "Why are you here?", shot_id="...", emotion="wary", index=0),
        make_beat("Tom", "I came to apologize.", index=1),
    ),
)
```

### Phase 2: Parametrize (set identity anchors)

Cast each character with a *canonical* face and voice. These anchors get
reused for every shot/beat that character appears in --- this is what gives
identity continuity.

```python
from falaw import cast_character, establish_environment

sarah = cast_character(
    "Sarah",
    "mid-30s, dark curly hair, wary eyes",
    reference_audio_url="https://.../sarah_sample.wav",
)
diner = establish_environment(
    "diner",
    "1950s chrome diner, neon outside, half-empty booths",
    time_of_day="midnight",
    lighting="cool fluorescents",
)
scene = scene.with_character(sarah).with_environment(diner)
```

### Phase 3: Render (caches; re-edits are cheap)

```python
from falaw import render_scene, save_scene

manifest = render_scene(scene)  # all beats + shots
save_scene(scene, "out/diner_v1.json")  # snapshot the IR alongside
```

### Phase 4: Direct (notes -> IR edits -> re-render)

```python
from falaw import apply_note_to_beat

# Pick the beat to direct.
beat = scene.beat("002-tom-...")
edited = apply_note_to_beat(beat, "He cracks on this line; tries to hide it.")
scene2 = scene.with_beat(edited)
manifest2 = render_scene(scene2)  # only the edited beat re-renders;
# the rest are cache hits.
```

For cross-cutting notes like "tighten the pacing", use
`apply_note_to_scene(scene, note)`. For non-LLM edits, just construct the
new dataclass yourself --- everything is frozen so `dataclasses.replace`
works.

### Local stitching

`falaw.local` (requires ffmpeg) stitches the per-beat lipsynced clips into
a watchable scene:

```python
from falaw.local import concatenate_clips

concatenate_clips(
    [m["url"] for m in manifest["beats"]], output_path="out/scene.mp4", transition_s=0.2
)
```

### Reference images and local files: falaw checks, you don't have to

falaw caches `url -> content hash`, but only a **fal** URL can be trusted
outright (fal mints one per upload, so it never re-points at other bytes).
Anything else --- a reference image on your own server, a `file://` clip you
re-rendered to the same path --- is **revalidated** before reuse:

```python
falaw.materialize_asset("https://mysite.com/reference.png")
# changed on the server? you get the new bytes and a new content hash.
# unchanged? one conditional request, no download.
```

You do **not** need to pass `refresh=True` for a mutable URL any more. If falaw
cannot check (the origin offers no `ETag`/`Last-Modified`, or you injected a
transport with no conditional-request support) it re-downloads rather than
trusting a hint it cannot verify. If the origin is *gone*, it falls back to the
bytes it already stored and warns.

Serving your own media from an immutable-by-construction store? Add the host so
falaw skips the check:

```python
from falaw.content import IMMUTABLE_URL_HOSTS

IMMUTABLE_URL_HOSTS.add("cdn.mysite.com")  # only if you mint URLs per upload
```

### The cache holds bytes now --- keep an eye on the disk

Since falaw#14 the cache content-addresses every result, so it stores the
**bytes** of every image, clip and audio track, not just JSON manifests. On a
machine that has rendered a real project that is gigabytes.

```python
import falaw

print(falaw.cache_usage().summary())  # per-area breakdown, largest first
```

Six areas, and only one of them is cheap to reclaim:

| area | dropping one costs | prunable |
|---|---|---|
| `assets` | **nothing**, while the blob survives (it is a *copy*) --- reclaim here first | yes |
| `content` | a **re-render**, for any call whose fal URL has since expired | yes |
| `manifests` | a **re-billed call**, unconditionally | yes |
| `url_index` | **the whole content area.** Blobs are reachable *only* through this index, so deleting it orphans them: expired-URL entries re-render while their bytes sit unreachable on disk | no |
| `scenes` | authored Scene IR --- lost work, not a re-render | no |
| `other` | unknown; falaw did not put it there | no |

So eviction is a spending decision, and nothing runs automatically. Every
prune is a dry run unless you say otherwise, and refuses to run unbounded
(or with a non-positive `older_than`, which is almost always a computed-zero
bug that would wipe the area):

```python
from datetime import timedelta

report = falaw.prune_assets(older_than=timedelta(days=30))  # dry run
print(report.summary())
falaw.prune_assets(older_than=timedelta(days=30), dry_run=False)
```

**Read the two warning numbers before passing `dry_run=False`:**

- `report.rebillable_entries` --- cache entries the prune puts back on the
  invoice (an upper bound).
- `report.unreferenced_candidates` --- blobs **no cache entry points at**.
  falaw cannot tell garbage from the last copy of a reference image you
  materialized, so it reports them rather than calling the prune free.

`report.freed_bytes` counts what was actually deleted, so a failed deletion
does not read as reclaimed space. `prune_content` and `prune_manifests` take
the same arguments; `max_bytes=N` evicts oldest-first until the area fits.

### execute_plan: three cache modes --- a re-run is not a cache bypass

`execute_plan(plan)` reads the cache and writes it. `execute_plan(plan, refresh=True)`
skips the read and **keeps the write** --- this is the one to use behind a "redo it"
or "re-verify this" switch. `execute_plan(plan, use_cache=False)` does neither, so it
throws away the result it just paid for and the next caller re-bills; reach for it only
when you truly want no cache interaction. Both together raises: nothing to key a write on.

### plan_llm_complete quotes a ceiling for the *routed* model

`fal-ai/any-llm` is one endpoint routing thirty models at two published request
tiers, so its single registry price under-quotes every premium model tenfold ---
including falaw's own default, `anthropic/claude-sonnet-4.5`. Planned LLM calls
price the `model` argument against `falaw/data/llm_rates.json` instead:

```python
from falaw import plan_llm_complete, llm_ceiling_usd

plan_llm_complete(prompt).estimated_cost_usd  # fal's request price for the model
plan_llm_complete(prompt, input_tokens=180_000, max_output_tokens=8_000)
# -> max(request price, upstream per-token cost of that prompt capped there)
```

Give both hints or neither. Half a bound is not a ceiling, so one hint alone ---
and a model the table does not price --- yields `estimated_cost_usd=None`, which
lights up `Plan.has_unknown_costs` and forces approval rather than quietly
quoting the router's flat price. `input_tokens` is what opts in: pass it and
`max_output_tokens` defaults to the `max_tokens` already in your `extra`; pass
neither and a capped call still quotes the request price. Hints are
estimator-only: adding one never moves the cache key or the plan hash.

The eager `llm_complete_with_receipt` prices the same way --- a receipt for a
premium routed model reads $0.01 with `cost_source="rate_table:per_call"`, so a
ledger that sums receipts and a gate that reads quotes agree.

Rates move. Every row carries its own `source` and `date` (`llm_rates.json`);
### A saved plan's price is a past fact --- re-quote it before you gate on it

Rates move, and 0.0.46 moved the LLM table **tenfold upward**. A plan you
persisted last month still carries last month's `estimated_cost_usd`, so
gating today's spend on it under-quotes --- the one direction a cost gate must
never err in. Every `plan_*` records how it priced each call in
`CallPlan.cost_basis` (which pricer, which rate table and version, and the
quantity hints --- a clip's `duration_s`, a prompt's token bounds --- that
never enter `arguments`), so the quote can be re-run:

```python
from falaw import plan_from_dict, reprice_plan

# frozen figures from plan time; re-quoting is pure data (no network, no cache peek)
out = reprice_plan(plan_from_dict(saved_row))

out.plan  # same calls, same plan_hash, today's costs
out.known_delta_usd  # how much the priced part moved
[(c.index, c.status, c.old_cost_usd, c.new_cost_usd) for c in out.changed]
```

Four per-call statuses: `unchanged`, `changed`, `unknown` (a basis that
today's tables cannot price --- a retired model, an unregistered pricer), and
`no_basis` (a plan saved before falaw#60, or built by something that recorded
nothing). **The last two clear the call's cost to `None`.** A stale number
re-presented as a current quote is the bug this exists to fix, so `no_basis`
is reported, never trusted --- read `out.unpriced` and refuse the gate, the
same judgement `Plan.has_unknown_costs` asks for.

`c.basis_changed` answers the audit question the frozen number never could:
did the price move because the *rate table* moved?

Quoted a call against your own `llm_rates=` table? Its basis says so, and
falaw's table will **not** re-quote it --- it comes back `unknown`. Two tables
are two sets of books, and re-pricing your reconciled $0.50 row at falaw's
published $0.01 would be a 50x under-quote wearing the clothes of a price
drop. Bring your own books instead:

```python
from falaw import DFLT_PRICERS, Pricer, llm_ceiling_usd
from falaw.llm_rates import CUSTOM_LLM_RATES_TABLE

mine = Pricer(
    quote=lambda b: llm_ceiling_usd(b.priced, rates=my_table, **b.quantities),
    table=CUSTOM_LLM_RATES_TABLE,  # must match the basis, or it is refused
    version=lambda: "2026-09",
)
reprice_plan(plan, pricers={**DFLT_PRICERS, "llm_rates": mine})
```

Rates move. Every row carries its own `source` and `date` (`llm_rates.json`);
a caller who has reconciled real numbers against their fal invoice passes their
own table as `llm_rates=`. `llm_rates.json` has no refresh job that runs
itself, so a quote is only as fresh as its last check --- run
`falaw.refresh_llm_rates()` (or `python -m falaw refresh-llm-rates`) to diff
it against fal's any-llm doc and OpenRouter's catalogue; it never overwrites
the committed table, only proposes one for a human to review.

## Read the journal first

Before novel work, glance at recent entries --- past sessions may have
left notes that save you time:

```python
from falaw import journal

for e in journal.recent(20):
    print(e.kind, "-", e.text[:120])
```

## Leave a journal entry when something surprises you

```python
from falaw import journal

journal.issue(
    "FLUX dev returned NSFW=True for a benign prompt",
    suggestion="Try guidance_scale=2.0",
    tags=("flux", "safety"),
)
journal.improvement(
    "Pass beat.emotion as a TTS prompt arg for emotion-aware models",
    tags=("backlog", "directorial"),
)
journal.note("schnell at quality='fast' returns 1024x1024 by default")
```

## Pick a model without memorizing IDs

```python
from falaw import list_models, pick_model

[m.id for m in list_models(category="image_to_video")]
pick_model(category="image_edit", quality_tier="ultra").id
```

## Tools

Every function below is a registered tool; bridges (MCP server, HTTP
service, UI) derive their surfaces from the same registry.
"""


_FOOTER_TEMPLATE = """
## Models known to falaw

The model registry lives at `falaw/data/models.json`. Refresh it from
`misc/docs/fal_ai_docs_full.md` when fal ships new models. Quick view:

```
{model_lines}
```

## When you can't find what you need

* Check `falaw/misc/docs/llms-full.txt` for a structured fal.ai overview.
* Check `falaw/misc/docs/fal_ai_docs_full.md` for the full corpus (~3MB).
* Drop into `falaw.call_fal(application, arguments)` for any model not
  yet wrapped --- this is the escape hatch. Then leave a `journal.improvement`
  asking for a proper tool wrapper.
"""


[docs] def build_skill_md(tools: Optional[Iterable[ToolSpec]] = None) -> str: """Render the SKILL.md content from the tool registry.""" tools = list(tools if tools is not None else list_tools()) parts = [_HEADER] for t in tools: parts.append(f"\n### `falaw.{t.name}`\n") parts.append(t.description.strip()) parts.append("") if t.examples: parts.append("Examples:") for ex in t.examples: parts.append(f" - `falaw.{t.name}(**{dict(ex)!r})`") parts.append("") model_lines = "\n".join( f" {m.category:20s} {m.quality_tier:10s} {m.id}" for m in sorted(list_models(), key=lambda x: (x.category, x.quality_tier)) ) parts.append(_FOOTER_TEMPLATE.format(model_lines=model_lines)) return "\n".join(parts)
[docs] def write_skill_files(target_dir: str) -> str: """Write SKILL.md (and a small references/ folder) under `target_dir`. Returns the path written. """ os.makedirs(os.path.join(target_dir, "references"), exist_ok=True) skill_path = os.path.join(target_dir, "SKILL.md") with open(skill_path, "w") as f: f.write(build_skill_md()) return skill_path