@hy-sde-org/dsh-graph-stream

@hy-sde-org/dsh-graph-stream

English | 中文

Summary

dsh-graph-stream is the derivation layer of the Agent Graph. It layers on top of @hy-sde-org/dsh-graph-control (the durable decision store) and owns everything that can be recomputed from committed rows: work-status projection, record folding, trace/route derivation, readiness intents, input handoffs, and the single-flight reconciliation driver (AgentGraphCoordinator) that walks Maka's authored drive loop (provision → supervise → select → render → execute) against that store.

The split follows Maka's one hard rule: the store is the authority, the stream layer never writes anything except through the store's own commit/claim/provision methods. Every projection here is deterministic and pure — recomputing it never starts work; admission and execution are store-sealed (claim with preallocated turn/run identity at the schedule revision it observed).

Ids are deterministic sha256 cut to 32 hex chars (graph_intent_…, graph_operator_…, graph_edge_…, graph_route_…, graph_record_…, graph_claim_…), so replays are idempotent by construction. Identity comparison uses UTF-16 code-unit order (compareAgentGraphIdentity), not locale comparison.

This package contributes no tool, prompt, or plugin row — the executor adapter and the supervisor tools consume it.

Table of Contents

• Use this package
• Understand the implementation
• Further Exploration
• Model Experience
• Known Limitations and Deferred Work

Use this package

import Storage, { storageBackendServiceKey } from '@deepseek-ai/dsh-storage'
import { GraphControlStore } from '@hy-sde-org/dsh-graph-control'
import { AgentGraphCoordinator } from '@hy-sde-org/dsh-graph-stream'

const backend = await ctx[storageBackendServiceKey('sqlite')]
const unit = await backend.kv.open(GraphControlStore.descriptor)
const store = await GraphControlStore.open(unit)

const coordinator = new AgentGraphCoordinator(graphId, {
  store,
  executor: { provisionOperator, runClaimedAgentGraphIntent, stopSession },
  recordSource,
  newId,
})

await coordinator.scheduleUpdate({ graphId, addWork: [work] }) // commits, then wakes the drive
const result = await coordinator.reconcileAndWait()             // runs one drive to idle

Understand the implementation

Deterministic identities and ordering

stableHash(value) = sha256: + hex of canonical JSON (sorted object keys, undefined/function/symbol → "[undefined]", bigint → numeric string, required/enum arrays sorted with localeCompare, Date → ISO). All ids use the first 32 hex chars. compareAgentGraphIdentity is UTF-16 code-unit order — stable across processes, unlike localeCompare.

Projections

• Schedule projection (projectAgentGraphSchedule): folds the append-only update log into the model-visible work view; validates contiguity (revisions from 1), no updates after a finish, no repeated work ids, wrong graph ids rejected. stopped wins over superseded.
• Record fold (readCommittedAgentGraphProjection): derives reference-only records from committed events per (operator, session); partial events skipped; at most one terminal record per activation; deterministic order.
• Trace (validateAgentGraphTraceTopology, buildAgentGraphTraceSnapshot): validates the DAG (duplicate ids/endpoints, self-loops, unknown operators, cycles via Kahn) and derives one route per (record × outgoing edge).
• Readiness (buildAgentGraphReadinessSnapshot): map policy — one intent per route received through a declared incoming edge, sealed against exactly the triggering records (policyFingerprint, readinessContextFingerprint, graph_intent_… via stable hashes).
• Handoff (hydrateAgentGraphInputHandoffs, renderAgentGraphScheduledWorkPrompt): resolves bounded conclusion text (16 KiB/record, 48 KiB total, … ellipsis, binary search over code points) and renders the operator prompt: instruction + GRAPH_OPERATOR_HANDOFF_PROTOCOL + `` with ` runClaimedAgentGraphIntent(input: AgentGraphRunClaimedIntentInput): Promise stopSession(sessionId: string, opts?: { reason?: string }): Promise }

The coordinator never calls a provider directly — the executor package supplies the subagent/worktree-backed implementation.

## Further Exploration

- [`@hy-sde-org/dsh-graph-control`](https://github.com/hy-sde/dsh-graph/blob/875d8f81d0e326396fdea488e41f1984677db539/packages/graph-control/README.md): the durable rows this layer folds.
- `src/reconcile.ts` / `src/coordinator.ts`: the drive loop and status derivation.
- `src/hash.ts` / `src/identity.ts`: the canonicalization and ordering primitives every id and fingerprint depends on.
- `tests/graph-stream.spec.ts` / `tests/reconcile.spec.ts`: projection, validation, handoff, and end-to-end drive scenarios against a real sqlite-backed store.

## Model Experience

The package is pure TypeScript with narrow, single-responsibility modules and no ambient state. Types are explicit; results are plain data (no class instances crossing module boundaries except the coordinator). Errors carry `reason` codes for validation failures and the store's conflict errors propagate unchanged.

## Known Limitations and Deferred Work

- Readiness policy kinds: `map` only — `all_settled` and supervisor-readiness kinds are deferred to the supervisor tool layer.
- No client projection/checkpointing (`onCheckpoint`) yet; no tool-view pagination; residency is a no-op.
- Map-policy intents are derived but not auto-dispatched by reconcile — they surface for the supervisor tools.
- Record shape is copy-with-provenance (slim), a deliberate deviation from Maka's 18-facet full record; the stream layer never mutates stored records.
- The coordinator is process-local: another process holding the same graph store will not wake this driver automatically (wake delivery is `dsh-graph-wakes`).