etl-state

1# SPDX-License-Identifier: Apache-2.02# yaml-language-server: $schema=https://nika.sh/spec/v1/workflow.schema.json3#4# TEMPLATE · etl-state · incremental data job with a state file.5#6#   [approve]──▶[fresh]──┬───────────────────────▶[save_state]   when approved7#    prompt      fetch   │                         nika:write8#   [previous_raw]─▶[previous]──┴──▶[delta]──when ≠∅──▶[process]9#    nika:read      nika:jq          json_diff          nika:jq10#11# The "only what changed since last time" job. A cursor file on disk remembers12# the last run, a fetch brings the current truth, and an RFC-6902 diff says13# exactly what moved — so downstream work is proportional to the CHANGE, not to14# the size of the dataset.15#16# Demonstrates ·17#   - a state file read→diff→write cycle · one path, both directions18#   - `nika:read` returns TEXT · it is parsed with `fromjson` before diffing19#     (skip that and the diff is meaningless — see `previous`)20#   - a blocking human gate that DOMINATES every path to the network and the21#     write · NEP-0002's Rule of Two, as a check22#   - `on_error: on_codes:` · forgive not-found ONLY, never a permission error23#24# TWO HONEST HINTS · this file ships RED-adjacent on purpose:25#   [headless-prompt] the gate is deliberately blocking · a `default:` here26#     completes the lethal trifecta and lights NIKA-SEC-009 (measured — add one27#     and watch the TRIFECTA rung flip). The hint is right that it needs a28#     human; that is the point.29#   [inputs] `previous_raw` reads a cursor file that does not exist yet · the30#     `on_error:` below handles exactly that, but `check` does not model31#     recovery. First run creates it.32#33# Needs · nothing to rehearse. For REAL use: a reachable source in34#   `const.source_url` — then delete the `on_error:` on `fresh`.35#36# Run · nika run <file> --output json37#     The run PAUSES at the gate (exit 4 · durable, not a failure) and prints38#     its own resume command. Answer it to continue.39nika: v140workflow:41  id: etl-state-template        # SLOT: kebab-case workflow id42  # SLOT: one honest sentence — `nika new` matches intents against these words43  description: "only what changed since last run · read state · fetch fresh · diff · process the delta · save state"44​45const:46  source_url: "https://api.example.com/v1/records"   # SLOT: the data source47  state_path: "./state/etl-state.json"               # SLOT: the cursor file48​49⛨permits:                            # the blast radius · default-deny once present50  ⛨tools: ["nika:fetch", "nika:jq", "nika:json_diff", "nika:prompt", "nika:read", "nika:write"]51  ⛨net: { http: ["api.example.com"] }    # SLOT: the source host from const.source_url52  ⛨fs:                               # the cursor file, read then rewritten — one path, both ways53    ⛨read: ["./state/etl-state.json"]    # SLOT: keep in step with const.state_path54    ⛨write: ["./state/etl-state.json"]   # SLOT: idem — the job writes nothing else55policy:                              # NEP-0014 · the gate runs under a NAMED endorsement mode56  endorsement: solo57​58tasks:59  # NEP-0002 · the Rule of Two, as a check. This run holds all three legs at60  # once: it reads a private file, ingests UNTRUSTED network content, and61  # persists that content into the very file the NEXT run reads as trusted62  # state. One human decision has to dominate EVERY path to that write — so the63  # gate sits before the first network touch, not next to the write (a gate the64  # fetch can route around dominates nothing). Blocking on purpose: a `default:`65  # here would disarm it, and the checker knows — measured, adding `default:`66  # flips TRIFECTA to `✖ NIKA-SEC-009 lethal trifecta complete`.67  approve:68    ◆invoke:69      tool: "nika:prompt"70      args:71        message: "Fetch ${{ const.source_url }} and persist it into ${{ const.state_path }}?"72​73  previous_raw:74    ◆invoke:75      tool: "nika:read"76      args: { path: "${{ const.state_path }}" }77    on_error:78      on_codes: [NIKA-BUILTIN-READ-001]   # not-found ONLY · a permission error still fails loudly79      recover: "[]"                       # first run · the STRING "[]", so `previous` parses it80                                          # exactly like a real file — one code path, not two81​82  previous:83    with:84      raw: ${{ tasks.previous_raw.output }}85    ◆invoke:86      # `nika:read` hands back TEXT, and `nika:json_diff` compares VALUES. Feed87      # it the raw string and the diff degrades to a single88      # `{"op":"replace","path":""}` carrying the entire new document — it looks89      # like it works, and it reports "everything changed" forever. Measured.90      # `fromjson` is what makes the next task an actual diff.91      tool: "nika:jq"92      args:93        input: "${{ with.raw }}"94        expression: "fromjson"95​96  fresh:97    with:98      go: ${{ tasks.approve.output }}   # the binding IS the edge · the gate dominates the fetch99    when: ${{ with.go == true }}        # declined → no network touch at all100    ◆invoke:101      tool: "nika:fetch"            # SLOT: fetch / read / exec · the fresh data102      args:103        url: "${{ const.source_url }}"104        mode: jq105        jq: ".records"106    on_error:107      recover: []                   # offline rehearsal · an empty batch, the delta stays quiet108​109  delta:110    with:                           # the bindings ARE the edges · previous + fresh → delta111      previous: ${{ tasks.previous.output }}112      fresh: ${{ tasks.fresh.output }}113    ◆invoke:114      tool: "nika:json_diff"        # RFC 6902 · empty patch = nothing new115      args:116        before: "${{ with.previous }}"117        after: "${{ with.fresh }}"118​119  process:120    with:121      delta: ${{ tasks.delta.output }}122    when: ${{ size(with.delta) > 0 }}123    ◆invoke:124      tool: "nika:jq"               # SLOT: the delta job (jq · infer · write…)125      args:126        input: "${{ with.delta }}"127        expression: "length"128​129  save_state:130    with:131      fresh: ${{ tasks.fresh.output }}132      go: ${{ tasks.approve.output }}   # the binding IS the edge · the gate dominates the write133    when: ${{ with.go == true }}        # a refusal is a VALUE · the write is skipped, never failed134    ◆invoke:135      tool: "nika:write"136      args:137        path: "${{ const.state_path }}"138        content: "${{ with.fresh }}"    # a VALUE · the engine serializes it, and the next run's139                                        # `fromjson` reads it straight back140        create_dirs: true141        overwrite: true142​143outputs:144  changes:145    value: ${{ tasks.delta.output }}146    description: "RFC 6902 ops since last run · empty = no-op run"147​