Synthetic Data Forge

---
name: synthetic-data-forge
description: >
  Generate realistic, referentially-consistent synthetic data from any schema — seed
  data, test fixtures, demo datasets, mock API payloads, load-test volume. Use on any
  request to populate a schema with believable values ("seed my database", "generate
  test data", "fake/sample/mock data", "fixtures for", "give me N example rows"). This
  is a SELF-BUILDING skill: it ships no executable code. On first use it directs the
  host agent to construct a local generator in its own sandbox from the spec below and
  prove it against a shipped acceptance test. Do NOT use to anonymize or transform REAL
  data (a privacy task), only to invent data from nothing.
spec_version: 2
---

# Synthetic Data Forge — self-building payload

This skill ships **markdown only**. It does not carry `forge.py` or any other runnable
file. Instead it carries (1) the durable procedure, (2) a Build Brief the host agent
follows to grow the generator in *its own* sandbox, and (3) an executable Acceptance
Harness that the generated tool must pass before it may be used. The code that runs was
authored by the host agent, in the user's environment, reviewable before it executes.
The spec is the invariant; the code is disposable and regenerated when `spec_version`
changes, so it never goes stale against library churn.

Distilled from two corroborated patterns: **schema-driven-synthetic-stream-generation**
(mock records from schema defaults + generator functions) and **schema-enforced-
generation** (validate every record against the schema before emit). Provenance:
TDengine, genkit.

## When to use
Whenever the deliverable is *volume of plausible data shaped like a schema*: seed/dev
data, test fixtures, demo datasets, mock API bodies, load-test volume, example pipeline
inputs. Not for perturbing or anonymizing real records.

## How this skill runs (the self-build loop)

On trigger, the host agent MUST:

1. **Check cache.** Look for a prior build at `./.forge_cache/v{spec_version}/forge_impl.py`.
   If present and `spec_version` is unchanged, reuse it — do not rebuild, do not spend
   tokens regenerating.
2. **Build if absent or stale.** If missing, or if `spec_version` is higher than the
   cached build, construct `forge_impl.py` per the Build Brief below, exposing exactly
   the mandated interface.
3. **Prove before use.** Write the Acceptance Harness verbatim to
   `.forge_cache/v{spec_version}/test_forge.py` and run it against the fresh build.
   **The build is not usable until the harness passes green.** If it fails, fix the
   implementation and re-run. Never emit data from an unproven build.
4. **Use.** Resolve the user's schema into the normalized form, call `forge(...)`, emit
   in the requested format, and report row counts, the seed, seeded edge cases, and any
   assumptions.

This loop is what makes the payload safe to ship as text: correctness is not "the agent
produced a file," it is "the agent's file passed the test that shipped with the spec."

## Build Brief

Construct a single module `forge_impl.py` exposing this exact interface — the Acceptance
Harness depends on it:

```
forge(schema: dict, seed: int) -> dict[str, list[dict]]
```

**Input — normalized schema dict:**
```
{
  "tables": [
    {
      "name": "users",
      "rows": 50,
      "pk": "id",
      "fields": [
        {"name": "id",        "type": "int",      "pk": true,  "unique": true},
        {"name": "email",     "type": "str",      "semantic": "email", "unique": true},
        {"name": "status",    "type": "str",      "enum": ["active","pending","suspended"]},
        {"name": "created_at","type": "datetime", "semantic": "created_at"},
        {"name": "note",      "type": "str",      "semantic": "text", "nullable": true}
      ]
    },
    {
      "name": "orders",
      "rows": 200,
      "pk": "id",
      "fields": [
        {"name": "id",      "type": "int",   "pk": true, "unique": true},
        {"name": "user_id", "type": "int",   "fk": "users.id"},
        {"name": "amount",  "type": "float", "semantic": "price"}
      ]
    }
  ]
}
```
Field keys: `name`, `type` (int|float|str|bool|datetime), optional `pk`, `unique`,
`nullable`, `enum` (list), `fk` ("table.column"), `semantic` (email|full_name|city|
price|count|status|created_at|updated_at|text|...).

**Output:** `{ "users": [ {field: value, ...}, ... ], "orders": [ ... ] }` — a list of
row dicts per table, in the order tables were declared.

**Invariants the implementation MUST hold (these are what the harness checks):**
1. Row counts match each table's `rows`.
2. `pk` fields are assigned sequentially (1..N) so they are stable, unique keys.
3. `unique` fields contain no duplicates within a table.
4. Every `fk` value resolves to a primary key that actually exists in the parent table.
   **No dangling foreign keys.** Generate parent tables first; draw child FKs only from
   the parent's realized PK set.
5. `enum` fields contain only allowed values.
6. Fields not marked `nullable` are never null; nullable fields may be null some of the
   time (a realistic minority), never always.
7. Determinism: same `(schema, seed)` produces byte-identical output across runs.
8. Semantic realism (loose): `email` values contain "@"; `created_at`/`updated_at` are
   ISO timestamps and `updated_at` (if present) is never before `created_at`.

**Generator guidance (realism, not just type-validity):**
- Prefer `faker` if available; fall back to stdlib `random` + small curated vocab lists.
- Derive dependent fields (email from a generated name; `updated_at` as a forward offset
  from `created_at`).
- Draw numerics from sensible distributions with domain bounds (log-normal-ish prices,
  small-count Poisson-ish quantities), never negative where impossible.
- Weight enum/status choices rather than uniform; give booleans a realistic base rate.
- Seed the RNG from `seed` and print it so a dataset can be reproduced.

## Acceptance Harness

Write this verbatim to `test_forge.py` beside the build and run it. It depends only on
the `forge(schema, seed)` contract, so it validates any conforming implementation. Green
is the only acceptable result.

```python
# test_forge.py — ships with the spec; the build must pass this before use.
from forge_impl import forge

SCHEMA = {
    "tables": [
        {"name": "users", "rows": 50, "pk": "id", "fields": [
            {"name": "id", "type": "int", "pk": True, "unique": True},
            {"name": "email", "type": "str", "semantic": "email", "unique": True},
            {"name": "status", "type": "str", "enum": ["active", "pending", "suspended"]},
            {"name": "created_at", "type": "datetime", "semantic": "created_at"},
            {"name": "note", "type": "str", "semantic": "text", "nullable": True},
        ]},
        {"name": "orders", "rows": 200, "pk": "id", "fields": [
            {"name": "id", "type": "int", "pk": True, "unique": True},
            {"name": "user_id", "type": "int", "fk": "users.id"},
            {"name": "amount", "type": "float", "semantic": "price"},
        ]},
    ]
}

def _index(schema):
    return {t["name"]: t for t in schema["tables"]}

def run():
    data = forge(SCHEMA, seed=42)
    tables = _index(SCHEMA)

    # 1. row counts
    for t in SCHEMA["tables"]:
        assert len(data[t["name"]]) == t["rows"], f"{t['name']} wrong row count"

    # 2/3. pk sequential + unique fields deduped
    for t in SCHEMA["tables"]:
        rows = data[t["name"]]
        for f in t["fields"]:
            if f.get("pk"):
                pks = [r[f["name"]] for r in rows]
                assert pks == list(range(1, len(rows) + 1)), f"{t['name']}.{f['name']} not sequential"
            if f.get("unique"):
                vals = [r[f["name"]] for r in rows]
                assert len(vals) == len(set(vals)), f"{t['name']}.{f['name']} has duplicates"

    # 4. no dangling foreign keys  <-- the load-bearing invariant
    for t in SCHEMA["tables"]:
        for f in t["fields"]:
            if f.get("fk"):
                ptab, pcol = f["fk"].split(".")
                pool = {r[pcol] for r in data[ptab]}
                for r in data[t["name"]]:
                    assert r[f["name"]] in pool, f"dangling FK {t['name']}.{f['name']}={r[f['name']]}"

    # 5. enum membership
    for t in SCHEMA["tables"]:
        for f in t["fields"]:
            if f.get("enum"):
                for r in data[t["name"]]:
                    assert r[f["name"]] in f["enum"], f"{t['name']}.{f['name']} bad enum {r[f['name']]}"

    # 6. non-nullable never null
    for t in SCHEMA["tables"]:
        for f in t["fields"]:
            if not f.get("nullable"):
                for r in data[t["name"]]:
                    assert r[f["name"]] is not None, f"{t['name']}.{f['name']} unexpected null"

    # 7. determinism
    assert forge(SCHEMA, seed=42) == data, "non-deterministic output for same seed"

    # 8. semantic realism (loose)
    for r in data["users"]:
        assert "@" in r["email"], "email missing @"

    total = sum(len(v) for v in data.values())
    print(f"PASS — {total} rows across {len(data)} tables, 0 dangling FKs")

if __name__ == "__main__":
    run()
```

## Notes
- **Versioning / anti-staleness:** bumping `spec_version` invalidates cached builds and
  forces re-derivation against current libraries. The version is also the cache key and
  the provenance anchor.
- **Scope:** this self-build model fits *skills* (procedures + modest generators). It is
  the wrong model for heavy capabilities (e.g. an FHE store) — those remain engineering
  projects, not agent-built-from-prose.
- **What "done" means:** the harness prints `PASS`. Nothing else counts as built.