zzstoatzz.io
This repository has no description
io.prefect.v0.*: versioned lexicons, transitions become events
go-crazy namespace decision: everything lives under io.prefect.v0.* and
problems down the road drive v1. seven lexicon files land in lexicons/
(defs with the shared #state shape, event, flow, flowRun, taskRun,
logBatch, workspace).
the server now writes state transitions as io.prefect.v0.event records
mirroring prefect's own Event schema — name prefect.{flow-run,task-run}.
{StateName}, resource labels, intended {from,to} + validated_state in
the payload, TID rkeys. the state collections are gone; current state
is appview-only, per docs/lexicons.md.
verified live: python client runs basic_flow to Completed and the repo
holds Pending->Running->Completed events with causal from-chains; zds
poc reruns clean with the v0 workspace space type and oplog.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
author
zzstoatzz
co-author
Claude Fable 5
date
(Jun 12, 2026, 2:30 AM -0500)
commit
387e639e
387e639e28b26680c7ff965f0c8394f18a3da6c9
parent
b15f1183
b15f1183f9936c3e4c51d3d32631308207e76742
+306
-54
+3
-3
README.md
+3
-3
README.md
···
42
42
just smoke
43
43
44
44
# then look at what it wrote
45
-
just records io.prefect.flowRunState # state history as at:// records
45
+
just records io.prefect.v0.event # the event log as at:// records
46
46
just export-car # the whole database, signed
47
47
```
48
48
···
53
53
## what works
54
54
55
55
- real `prefect>=3.0` python client runs flows to completion
56
-
- flows, flow runs, task runs, state transitions, log batches as records
57
-
under `io.prefect.*` collections
56
+
- flows, flow runs, task runs as entity records; state transitions as
57
+
`io.prefect.v0.event` records (see `docs/lexicons.md` and `lexicons/`)
58
58
- signed v3 commits with monotonic TID revs; CAR persisted atomically on
59
59
every commit; verified round-trip in tests via `zat.verifyCommitCar`
60
60
- `com.atproto.sync.getRepo`, `com.atproto.repo.describeRepo`,
+2
-2
justfile
+2
-2
justfile
···
24
24
curl -s http://localhost:4200/xrpc/com.atproto.sync.getRepo -o {{out}}
25
25
@echo "wrote {{out}} — verify offline against the workspace did"
26
26
27
-
# list records in a collection (e.g. just records io.prefect.flowRunState)
28
-
records collection="io.prefect.flowRunState":
27
+
# list records in a collection (e.g. just records io.prefect.v0.event)
28
+
records collection="io.prefect.v0.event":
29
29
curl -s 'http://localhost:4200/xrpc/com.atproto.repo.listRecords?collection={{collection}}' | jq .
30
30
31
31
# wipe local state (signing key + repo CAR)
+37
lexicons/io.prefect.v0.defs.json
+37
lexicons/io.prefect.v0.defs.json
···
1
+
{
2
+
"lexicon": 1,
3
+
"id": "io.prefect.v0.defs",
4
+
"defs": {
5
+
"state": {
6
+
"type": "object",
7
+
"description": "a snapshot of a run state. one shape serves flow runs and task runs, mirroring prefect's State model.",
8
+
"required": ["type", "timestamp"],
9
+
"properties": {
10
+
"type": {
11
+
"type": "string",
12
+
"knownValues": [
13
+
"SCHEDULED",
14
+
"PENDING",
15
+
"RUNNING",
16
+
"COMPLETED",
17
+
"FAILED",
18
+
"CANCELLED",
19
+
"CANCELLING",
20
+
"CRASHED",
21
+
"PAUSED"
22
+
]
23
+
},
24
+
"name": { "type": "string", "maxLength": 256 },
25
+
"timestamp": { "type": "string", "format": "datetime" },
26
+
"message": { "type": "string", "maxLength": 4096 },
27
+
"stateDetails": { "type": "unknown" }
28
+
}
29
+
},
30
+
"labels": {
31
+
"type": "object",
32
+
"description": "open label set describing a resource. by convention carries prefect.resource.id and optionally prefect.resource.name / prefect.resource.role.",
33
+
"required": [],
34
+
"properties": {}
35
+
}
36
+
}
37
+
}
+51
lexicons/io.prefect.v0.event.json
+51
lexicons/io.prefect.v0.event.json
···
1
+
{
2
+
"lexicon": 1,
3
+
"id": "io.prefect.v0.event",
4
+
"defs": {
5
+
"main": {
6
+
"type": "record",
7
+
"description": "something that happened to a resource. the only append-only observability shape: state transitions, lifecycle changes, and custom events all land here. a TID-keyed collection scan is the chronological event log.",
8
+
"key": "tid",
9
+
"record": {
10
+
"type": "object",
11
+
"required": ["occurred", "event", "resource"],
12
+
"properties": {
13
+
"occurred": {
14
+
"type": "string",
15
+
"format": "datetime",
16
+
"description": "when the event happened from the sender's perspective"
17
+
},
18
+
"event": {
19
+
"type": "string",
20
+
"maxLength": 256,
21
+
"description": "dotted kebab-case event name, e.g. prefect.flow-run.Completed"
22
+
},
23
+
"resource": {
24
+
"type": "ref",
25
+
"ref": "io.prefect.v0.defs#labels",
26
+
"description": "the primary resource this event concerns"
27
+
},
28
+
"related": {
29
+
"type": "array",
30
+
"items": { "type": "ref", "ref": "io.prefect.v0.defs#labels" },
31
+
"description": "additional resources involved in this event"
32
+
},
33
+
"payload": {
34
+
"type": "unknown",
35
+
"description": "open-ended data describing what happened; state transitions carry intended/validated_state shaped by io.prefect.v0.defs#state"
36
+
},
37
+
"eventId": {
38
+
"type": "string",
39
+
"maxLength": 64,
40
+
"description": "client-provided identifier (uuid7), for dedup and causal references"
41
+
},
42
+
"follows": {
43
+
"type": "string",
44
+
"maxLength": 64,
45
+
"description": "eventId of an event known to have occurred before this one"
46
+
}
47
+
}
48
+
}
49
+
}
50
+
}
51
+
}
+23
lexicons/io.prefect.v0.flow.json
+23
lexicons/io.prefect.v0.flow.json
···
1
+
{
2
+
"lexicon": 1,
3
+
"id": "io.prefect.v0.flow",
4
+
"defs": {
5
+
"main": {
6
+
"type": "record",
7
+
"description": "a flow definition. control-plane entity: identity and configuration, no live state.",
8
+
"key": "any",
9
+
"record": {
10
+
"type": "object",
11
+
"required": ["name"],
12
+
"properties": {
13
+
"name": { "type": "string", "maxLength": 512 },
14
+
"created": { "type": "string", "format": "datetime" },
15
+
"tags": {
16
+
"type": "array",
17
+
"items": { "type": "string", "maxLength": 256 }
18
+
}
19
+
}
20
+
}
21
+
}
22
+
}
23
+
}
+38
lexicons/io.prefect.v0.flowRun.json
+38
lexicons/io.prefect.v0.flowRun.json
···
1
+
{
2
+
"lexicon": 1,
3
+
"id": "io.prefect.v0.flowRun",
4
+
"defs": {
5
+
"main": {
6
+
"type": "record",
7
+
"description": "a flow run. orchestration-plane entity: identity and configuration only — live state is appview territory, derived from io.prefect.v0.event records.",
8
+
"key": "any",
9
+
"record": {
10
+
"type": "object",
11
+
"required": ["flow"],
12
+
"properties": {
13
+
"flow": {
14
+
"type": "string",
15
+
"format": "at-uri",
16
+
"description": "the io.prefect.v0.flow this run executes"
17
+
},
18
+
"name": { "type": "string", "maxLength": 512 },
19
+
"created": { "type": "string", "format": "datetime" },
20
+
"parameters": { "type": "unknown" },
21
+
"parametersBlob": {
22
+
"type": "blob",
23
+
"accept": ["application/json"],
24
+
"maxSize": 10000000,
25
+
"description": "parameters above the inline threshold"
26
+
},
27
+
"tags": {
28
+
"type": "array",
29
+
"items": { "type": "string", "maxLength": 256 }
30
+
},
31
+
"deployment": { "type": "string", "format": "at-uri" },
32
+
"parentTaskRun": { "type": "string", "format": "at-uri" },
33
+
"idempotencyKey": { "type": "string", "maxLength": 512 }
34
+
}
35
+
}
36
+
}
37
+
}
38
+
}
+27
lexicons/io.prefect.v0.logBatch.json
+27
lexicons/io.prefect.v0.logBatch.json
···
1
+
{
2
+
"lexicon": 1,
3
+
"id": "io.prefect.v0.logBatch",
4
+
"defs": {
5
+
"main": {
6
+
"type": "record",
7
+
"description": "a batch of log lines as shipped by the client. kept apart from io.prefect.v0.event: different volume class, different retention.",
8
+
"key": "tid",
9
+
"record": {
10
+
"type": "object",
11
+
"required": [],
12
+
"properties": {
13
+
"entries": {
14
+
"type": "unknown",
15
+
"description": "inline log entries below the blob threshold"
16
+
},
17
+
"entriesBlob": {
18
+
"type": "blob",
19
+
"accept": ["application/json"],
20
+
"maxSize": 50000000
21
+
},
22
+
"flowRun": { "type": "string", "format": "at-uri" }
23
+
}
24
+
}
25
+
}
26
+
}
27
+
}
+33
lexicons/io.prefect.v0.taskRun.json
+33
lexicons/io.prefect.v0.taskRun.json
···
1
+
{
2
+
"lexicon": 1,
3
+
"id": "io.prefect.v0.taskRun",
4
+
"defs": {
5
+
"main": {
6
+
"type": "record",
7
+
"description": "a task run within a flow run. identity and configuration only — live state is derived from io.prefect.v0.event records.",
8
+
"key": "any",
9
+
"record": {
10
+
"type": "object",
11
+
"required": ["taskKey", "dynamicKey"],
12
+
"properties": {
13
+
"flowRun": {
14
+
"type": "string",
15
+
"format": "at-uri",
16
+
"description": "the io.prefect.v0.flowRun this task ran within"
17
+
},
18
+
"name": { "type": "string", "maxLength": 512 },
19
+
"created": { "type": "string", "format": "datetime" },
20
+
"taskKey": { "type": "string", "maxLength": 512 },
21
+
"dynamicKey": { "type": "string", "maxLength": 512 },
22
+
"cacheKey": { "type": "string", "maxLength": 2048 },
23
+
"cacheExpiration": { "type": "string", "format": "datetime" },
24
+
"taskInputs": { "type": "unknown" },
25
+
"tags": {
26
+
"type": "array",
27
+
"items": { "type": "string", "maxLength": 256 }
28
+
}
29
+
}
30
+
}
31
+
}
32
+
}
33
+
}
+19
lexicons/io.prefect.v0.workspace.json
+19
lexicons/io.prefect.v0.workspace.json
···
1
+
{
2
+
"lexicon": 1,
3
+
"id": "io.prefect.v0.workspace",
4
+
"defs": {
5
+
"main": {
6
+
"type": "record",
7
+
"description": "the permissioned-space type for a prefect workspace (com.atproto.space.* on zds). all io.prefect.v0.* records of a private workspace are written into a space of this type; writer credentials are minted by the workspace owner.",
8
+
"key": "any",
9
+
"record": {
10
+
"type": "object",
11
+
"required": [],
12
+
"properties": {
13
+
"name": { "type": "string", "maxLength": 512 },
14
+
"description": { "type": "string", "maxLength": 4096 }
15
+
}
16
+
}
17
+
}
18
+
}
19
+
}
+11
-7
scripts/devlog-poc
+11
-7
scripts/devlog-poc
···
27
27
HANDLE = os.environ.get("ATPROTO_DEVLOG_HANDLE", "zzstoatzzdevlog.bsky.social")
28
28
PASSWORD = os.environ["ATPROTO_DEVLOG_PASSWORD"]
29
29
30
-
COLLECTION = "io.prefect.flowRunState"
30
+
COLLECTION = "io.prefect.v0.event"
31
31
32
32
33
33
def now() -> str:
···
48
48
49
49
record = {
50
50
"$type": COLLECTION,
51
-
"runId": "demo-run-0001",
52
-
"type": "COMPLETED",
53
-
"name": "Completed",
54
-
"timestamp": now(),
55
-
"message": "demo: orchestration state on a stock bluesky pds",
51
+
"occurred": now(),
52
+
"event": "prefect.flow-run.Completed",
53
+
"resource": {
54
+
"prefect.resource.id": "prefect.flow-run.demo-run-0001",
55
+
"prefect.resource.name": "demo",
56
+
},
57
+
"payload": {
58
+
"validated_state": {"type": "COMPLETED", "name": "Completed", "timestamp": now()},
59
+
},
56
60
}
57
61
58
62
async with websockets.connect(
···
75
79
print(f" did: {event.get('did')}")
76
80
print(f" collection: {commit.get('collection')}")
77
81
print(f" rkey: {commit.get('rkey')}")
78
-
print(f" state: {commit.get('record', {}).get('type')}")
82
+
print(f" event: {commit.get('record', {}).get('event')}")
79
83
80
84
listed = (
81
85
await http.get(
+18
-14
scripts/zds-poc
+18
-14
scripts/zds-poc
···
26
26
HANDLE = os.environ.get("ATPROTO_WAOW_HANDLE", "waow.tech")
27
27
PASSWORD = os.environ["ATPROTO_WAOW_PASSWORD"]
28
28
29
-
SPACE_TYPE = "io.prefect.workspace"
29
+
SPACE_TYPE = "io.prefect.v0.workspace"
30
30
SPACE_KEY = "demo-workspace"
31
-
STATE_COLLECTION = "io.prefect.flowRunState"
31
+
EVENT_COLLECTION = "io.prefect.v0.event"
32
32
33
33
34
34
def now() -> str:
···
62
62
http, "com.atproto.repo.createRecord", headers=auth,
63
63
body={
64
64
"repo": did,
65
-
"collection": "io.prefect.flow",
65
+
"collection": "io.prefect.v0.flow",
66
66
"record": {
67
-
"$type": "io.prefect.flow",
67
+
"$type": "io.prefect.v0.flow",
68
68
"name": "hello-zds",
69
69
"created": now(),
70
70
},
···
97
97
body={
98
98
"space": space_uri,
99
99
"repo": did,
100
-
"collection": STATE_COLLECTION,
100
+
"collection": EVENT_COLLECTION,
101
101
"record": {
102
-
"$type": STATE_COLLECTION,
103
-
"runId": "private-run-0001",
104
-
"type": "COMPLETED",
105
-
"name": "Completed",
106
-
"timestamp": now(),
107
-
"message": "private orchestration state in a permissioned space",
102
+
"$type": EVENT_COLLECTION,
103
+
"occurred": now(),
104
+
"event": "prefect.flow-run.Completed",
105
+
"resource": {
106
+
"prefect.resource.id": "prefect.flow-run.private-run-0001",
107
+
"prefect.resource.name": "private-demo",
108
+
},
109
+
"payload": {
110
+
"validated_state": {"type": "COMPLETED", "name": "Completed", "timestamp": now()},
111
+
},
108
112
},
109
113
},
110
114
)
111
115
state.raise_for_status()
112
-
print(f"[private] state record: {state.json()['uri']}")
116
+
print(f"[private] event record: {state.json()['uri']}")
113
117
114
118
listed = xrpc(
115
119
http, "com.atproto.space.listRecords", headers=auth,
116
-
params={"space": space_uri, "repo": did, "collection": STATE_COLLECTION},
120
+
params={"space": space_uri, "repo": did, "collection": EVENT_COLLECTION},
117
121
)
118
122
listed.raise_for_status()
119
123
n = len(listed.json().get("records", []))
···
132
136
# ---- the privacy proof: same read, no auth ---------------------------
133
137
anon = xrpc(
134
138
http, "com.atproto.space.listRecords",
135
-
params={"space": space_uri, "repo": did, "collection": STATE_COLLECTION},
139
+
params={"space": space_uri, "repo": did, "collection": EVENT_COLLECTION},
136
140
)
137
141
if anon.status_code in (401, 403):
138
142
print(f"[private] unauthenticated read rejected: {anon.status_code} {anon.json().get('error', '')}")
+44
-28
src/api.zig
+44
-28
src/api.zig
···
18
18
19
19
const log = std.log.scoped(.api);
20
20
21
-
const flow_collection = "io.prefect.flow";
22
-
const flow_run_collection = "io.prefect.flowRun";
23
-
const flow_run_state_collection = "io.prefect.flowRunState";
24
-
const task_run_collection = "io.prefect.taskRun";
25
-
const task_run_state_collection = "io.prefect.taskRunState";
26
-
const log_batch_collection = "io.prefect.logBatch";
21
+
const flow_collection = "io.prefect.v0.flow";
22
+
const flow_run_collection = "io.prefect.v0.flowRun";
23
+
const task_run_collection = "io.prefect.v0.taskRun";
24
+
const event_collection = "io.prefect.v0.event";
25
+
const log_batch_collection = "io.prefect.v0.logBatch";
27
26
28
27
pub const Api = struct {
29
28
allocator: std.mem.Allocator,
···
282
281
self.lock.lock(self.io) catch return error.LockFailed;
283
282
defer self.lock.unlock(self.io);
284
283
285
-
try self.applyInitialState(arena, &row, parsed, now);
284
+
try self.applyInitialState(arena, &row, parsed, now, .flow_run);
286
285
287
286
const record_json = try std.fmt.allocPrint(arena,
288
287
\\{{"$type":"{s}","id":"{s}","created":"{s}","name":"{s}","flowId":"{s}","parameters":{s},"tags":{s}}}
···
325
324
self.lock.lock(self.io) catch return error.LockFailed;
326
325
defer self.lock.unlock(self.io);
327
326
328
-
try self.applyInitialState(arena, &row, parsed, now);
327
+
try self.applyInitialState(arena, &row, parsed, now, .task_run);
329
328
330
329
const record_json = try std.fmt.allocPrint(arena,
331
330
\\{{"$type":"{s}","id":"{s}","created":"{s}","name":"{s}","flowRunId":{s},"taskKey":"{s}","dynamicKey":"{s}"}}
···
344
343
}
345
344
346
345
/// runs are created with an optional initial state (the client sends
347
-
/// Pending). records it like any other transition.
348
-
fn applyInitialState(self: *Api, arena: std.mem.Allocator, row: *RunRow, parsed: std.json.Value, now: []const u8) !void {
346
+
/// Pending). recorded like any other transition: an event.
347
+
fn applyInitialState(self: *Api, arena: std.mem.Allocator, row: *RunRow, parsed: std.json.Value, now: []const u8, kind: RunKind) !void {
349
348
const state = parsed.object.get("state") orelse return;
350
349
if (state != .object) return;
351
350
···
362
361
row.state_timestamp = now;
363
362
row.raw_state_details = try rawField(arena, state, "state_details", "{}");
364
363
365
-
const is_task = row.task_key.len > 0;
366
-
try self.writeStateRecord(arena, if (is_task) task_run_state_collection else flow_run_state_collection, row.*, null);
364
+
try self.writeTransitionEvent(arena, kind, row.*, null);
367
365
}
368
366
369
367
fn setState(self: *Api, req: *httpz.Request, res: *httpz.Response, id: []const u8, kind: RunKind) !void {
···
403
401
else
404
402
null;
405
403
404
+
const prior_type: ?[]const u8 = if (row.state_id.len > 0)
405
+
try arena.dupe(u8, row.state_type)
406
+
else
407
+
null;
408
+
406
409
row.state_id = state_id;
407
410
row.state_type = try arena.dupe(u8, state_type);
408
411
row.state_name = try arena.dupe(u8, state_name);
···
416
419
}
417
420
if (isTerminal(state_type)) row.end_time = now;
418
421
419
-
try self.writeStateRecord(arena, switch (kind) {
420
-
.flow_run => flow_run_state_collection,
421
-
.task_run => task_run_state_collection,
422
-
}, row, message);
422
+
try self.writeTransitionEvent(arena, kind, row, prior_type);
423
423
424
424
try self.storeRow(arena, kind_str, row);
425
425
···
442
442
res.body = out.written();
443
443
}
444
444
445
-
/// append-only state transition record, rkey = TID (chronological)
446
-
fn writeStateRecord(self: *Api, arena: std.mem.Allocator, collection: []const u8, row: RunRow, message: ?[]const u8) !void {
445
+
/// state transitions are io.prefect.v0.event records — there is no
446
+
/// state collection. mirrors prefect's own convention: event name
447
+
/// `prefect.flow-run.{StateName}`, resource labels, state in payload.
448
+
fn writeTransitionEvent(self: *Api, arena: std.mem.Allocator, kind: RunKind, row: RunRow, prior_type: ?[]const u8) !void {
449
+
const entity = switch (kind) {
450
+
.flow_run => "flow-run",
451
+
.task_run => "task-run",
452
+
};
447
453
const record_json = try std.fmt.allocPrint(arena,
448
-
\\{{"$type":"{s}","id":"{s}","runId":"{s}","type":"{s}","name":"{s}","timestamp":"{s}","message":{s},"stateDetails":{s}}}
454
+
\\{{"$type":"{s}","occurred":"{s}","event":"prefect.{s}.{s}","resource":{{"prefect.resource.id":"prefect.{s}.{s}","prefect.resource.name":"{s}"}},"payload":{{"intended":{{"from":{s},"to":"{s}"}},"validated_state":{{"type":"{s}","name":"{s}","timestamp":"{s}","message":{s},"stateDetails":{s}}}}},"eventId":"{s}"}}
449
455
, .{
450
-
collection, row.state_id,
451
-
row.id, row.state_type,
452
-
row.state_name, row.state_timestamp,
453
-
try jsonOptString(arena, message), row.raw_state_details,
456
+
event_collection,
457
+
row.state_timestamp,
458
+
entity,
459
+
row.state_name,
460
+
entity,
461
+
row.id,
462
+
row.name,
463
+
try jsonOptString(arena, prior_type),
464
+
row.state_type,
465
+
row.state_type,
466
+
row.state_name,
467
+
row.state_timestamp,
468
+
try jsonOptString(arena, row.state_message),
469
+
row.raw_state_details,
470
+
row.state_id,
454
471
});
455
472
456
-
// rkey: reuse the commit rev clock for a fresh TID so state records
457
-
// sort chronologically within their collection
473
+
// rkey: a fresh TID past the commit rev clock so events sort
474
+
// chronologically within the collection
458
475
const now_us: u64 = @intCast(@max(0, util.nowMicros(self.io)));
459
476
var tid = zat.Tid.fromTimestamp(@max(now_us, self.repo.last_rev_us + 1), 1);
460
477
const rkey = try arena.dupe(u8, tid.str());
461
-
try self.writeRecord(arena, collection, rkey, record_json);
478
+
try self.writeRecord(arena, event_collection, rkey, record_json);
462
479
}
463
480
464
481
fn getFlowRun(self: *Api, req: *httpz.Request, res: *httpz.Response, id: []const u8) !void {
···
646
663
.collections = &[_][]const u8{
647
664
flow_collection,
648
665
flow_run_collection,
649
-
flow_run_state_collection,
650
666
task_run_collection,
651
-
task_run_state_collection,
667
+
event_collection,
652
668
log_batch_collection,
653
669
},
654
670
.handleIsCorrect = false,