pjf

Pipe Journal Fabric.

File-backed, replayable Unix pipes with a small streaming DSL on top. A topic is a file, a record is a line, a topology is a YAML of supervised bash -c pipelines.

For the why, see PHILOSOPHY.md.

Install

git clone https://github.com/99-not-out/pjf.git
cd pjf
export PATH="$PWD/bin:$PATH"
export PJF_HOME="$PWD"    # defaults to $PWD; point elsewhere if needed

Requires: bash 4+, awk, sed, jq (only for the LLM agent examples), curl (same). No other dependencies.

Quickstart

# Produce some records into a topic file.
printf 'category=food\tamount=12\n' | pjf-produce expenses.kfmt
printf 'category=transport\tamount=15\n' | pjf-produce expenses.kfmt

# Consume the whole thing:
pjf-consume expenses.kfmt        # prints all lines, tail -F style

# Or as a group with durable offsets:
pjf-consume expenses.kfmt my_group

Run the examples:

examples/demo-tally.sh               # reduce + journal durability
examples/demo-logs.sh                # raw access.log -> parsed -> fan-out
PJF_MOCK=1 examples/demo-debate.sh # 6-agent LLM pipeline (mock; set
                                      # ANTHROPIC_API_KEY for real calls)

Run the test suite:

tests/run.sh                          # 17 tests, ~15 seconds

When to use it

Plain Unix pipes are great for one live computation:

• one producer
• one process chain
• one run
• one reader
• no restarts in the middle

pjf is useful when you still want pipe-style composition, but need a small local event fabric instead of one fragile process chain:

• one source feeding multiple independent consumers
• one stage restarting without rebuilding the whole topology
• inspectable intermediate streams (cat topics/foo.kfmt)
• producers and consumers living on different timelines
• optional replay and restartable state
• cross-language or human-in-the-loop coordination

Good fits

• local log routing and triage
• CLI-driven ETL and data cleanup pipelines
• event-driven shell automation on one machine
• webhook, scrape, or file-watch ingestion into local workflows
• rolling status/metrics aggregation via durable reduce
• multi-process test harnesses and dev orchestration
• human review queues where the topic file is the audit trail
• agent coordination, which was the original motivating case

Why not just use pipes?

The missing pieces in ordinary pipes are mostly about lifecycle and multiplexing:

• pipes are usually one reader; fan-out takes awkward tee trees
• intermediates disappear unless explicitly captured
• restarting one stage often means rebuilding the whole chain
• producers and consumers must be alive at the same time
• there is no natural replay point or durable per-consumer position

With pjf, the file-backed topic is the buffer, the inspection point, and optionally the recovery point.

Core concepts

Records are kv-lines. TAB-separated key=value fields, one record per line. Values can contain anything except TAB and newline (the writer strips those at format boundaries).

id=42	user=alice	amount=12

Topics are files. topics/<name>.kfmt for structured (kv-line) topics. Anything else is treated as raw (one value per line). The .kfmt suffix is a human convention — pjf doesn't enforce it — but the DSL uses it to reason about where to cross format boundaries.

A topology is a YAML. Each pipeline is a source + ops + a sink:

my_pipeline:
  from: events.kfmt
  transform: '@if ((kv["amount"] + 0) > 10) kv_emit()'
  to: big_events.kfmt

Run it with pjf-run topology.yml.

DSL reference

Key	Meaning
from: <topic>	Source file to tail from
to: <topic>	Sink file to append to (required)
group: <name>	Consumer group — defaults to pipeline name, tracks offsets
transform: @<awk>	Stateless awk body. kv[] pre-parsed, kv_emit() helper
reduce: <key> @<awk>	Stateful, journal-backed. state[], kv[], emit()
wrap: <key>	Raw line → <key>=<line> (kv record)
unwrap: <key>	kv record → value of <key> (raw line)
action: <path> [args]	Insert an arbitrary stdin/stdout executable

Exactly one from and one to per pipeline. reduce must be the last op in a pipeline (if present).

Worked examples

Stateless filter + projection:

errors:
  from: access.kfmt
  transform: '@if ((kv["status"]+0) >= 400) { delete kv["body"]; kv_emit() }'
  to: errors.kfmt

Stateful counting (running counts per category, durable across restarts):

tally:
  from: events.kfmt
  reduce: category @{ state["count"] = (state["count"] + 0) + 1; emit() }
  to: tallies.kfmt

The sink tallies.kfmt is the state journal. On restart, the reducer scans it for last-per-key and resumes the consumer from the highest committed __offset.

Fan-out (two consumers of the same source, different ops):

evens:
  from: numbers.kfmt
  group: evens
  transform: '@if ((kv["n"]+0) % 2 == 0) kv_emit()'
  to: evens.kfmt

odds:
  from: numbers.kfmt
  group: odds
  transform: '@if ((kv["n"]+0) % 2 == 1) kv_emit()'
  to: odds.kfmt

Two pipelines, two groups — each reads every record independently with its own offset file. This is how you get broadcast, conditional routing, and mirror-for-archive patterns. There is deliberately no branch: or tee: specialist op.

Format boundary (ingest a raw log file, parse to kv):

parse:
  from: access.log
  action: bin/parse-weblog
  to: access.kfmt

Where bin/parse-weblog is any stdin-to-stdout script that reads raw lines and writes kv-lines.

Reducer contract

Reducers receive tagged records carrying __offset=N. For each record:

1. kv[] is populated from the incoming record's fields.
2. state[] is loaded from the reducer's per-key in-memory accumulator.
3. Your awk body runs. Mutate state[] freely; call emit() to append a journal row.
4. state[] is saved back.

Call emit() on every input. The reducer's sink file is its state journal — emit() writes a row containing __offset, the group key's value, and every state[] field. The in-memory committed offset only advances inside emit(). If your reducer buffers partial state and emits later, buffered-but-not-journaled state is lost on restart.

Downstream consumers filter/dedupe the resulting state-snapshot stream. The examples show a file-marker dedupe pattern for agents that should process each id exactly once.

CLI scripts

Script	Purpose
pjf-produce <topic>	Append stdin to a topic file
pjf-consume <topic> [group]	Tail a topic with optional offset tracking
pjf-transform '@<awk>'	Apply an awk body per record with kv[]/kv_emit()
pjf-reduce <key> '@<awk>' [journal]	Stateful per-key reduce
pjf-wrap <key> / pjf-unwrap <key>	Raw ↔ kv format boundary
pjf-run <topology.yml>	Compile and supervise a topology
pjf-graph [-d] <topology.yml>	ASCII tree (or DOT) of the DAG
kv-get / kv-set / kv-drop / kv-strip	Per-field helpers for agent scripts

Layout on disk

topics/<name>.kfmt     kv-line topic files (structured)
topics/<name>.log      raw topic files (any suffix, or none)
offsets/<group>__<topic>   committed offset per consumer group
state/*                agent-private state (dedup markers, etc.)

Limits

• Single node.
• At-least-once — expect duplicates at crash boundaries.
• Throughput is in the thousands of records/sec, not millions.
• No schema enforcement, no ACL.
• Reducer journals grow linearly — compact offline with an awk last-per-key pass when they get big.

When not to use it

• distributed systems
• exactly-once requirements
• sustained high-throughput streaming
• untrusted or multi-tenant environments
• cases where a single ordinary pipe chain is already enough

See PHILOSOPHY.md for the design reasoning behind these choices and the ops that were deliberately rejected.