Immutable source bytes · bounded boundary recovery · normalized observations · deterministic evidence · replayable TDS calls
Compatibility baseline: Staqtapp-TDS v3.5.2 Programmer Core API Guide.
Image status: the header image is concept art, not an implemented feature or benchmark. Replace the placeholder URL before publishing.
This demonstration records how an uncertain byte region becomes a reviewable trajectory observation and a durable TDS evidence episode.
PYPI: https://pypi.org/project/staqtapp-tds/
It separates five different things that must not be conflated:
- the immutable sensor recording;
- the recursive search for record boundaries;
- the normalized physical observation parsed from the selected record;
- the ordered public TDS calls and their results;
- an independently verified label used for training or evaluation.
The result is suitable for replay, audit, dataset construction, and learning correct TDS API sequences. TDS stores and ranks evidence; it does not decide whether a trajectory is physically true and it does not train a model.
This is an evidence and telemetry architecture.
It provides:
- bounded recovery of uncertain byte regions;
- candidate and plugin evidence;
- normalized, versioned trajectory observations;
- deterministic candidate ranking for fixed inputs and fixed state;
- public-API-only TDS storage, verification, persistence, and reads;
- replay and drift evidence;
- model-ready examples with explicit label provenance.
It does not provide engagement authority, target selection, aiming, fire control, or autonomous control behavior. Diagnostics observe copied events and snapshots; they never control extraction or storage locks.
flowchart TD
A["Immutable recording bytes"] --> B["Bounded candidate graph"]
B --> C["Selected region"]
C --> D["Normalized observation"]
D --> E["TDS evidence episode"]
E --> F["Replay or training export"]
E -. "copied snapshots" .-> G["Diagnostics"]
The lifecycle is intentionally one-way:
- snapshot mutable input into immutable
bytes; - hash and store the source identity;
- evaluate approximate
[start, end_exclusive)requests with bounded recursion; - store the complete candidate graph and ranking evidence;
- copy only the selected payload at the persistence boundary;
- parse that payload into a normalized observation with units and a coordinate frame;
- record public TDS calls and stable results;
- verify the filesystem, then flush each TDS node with atomic replacement;
- export only committed, verified episodes for replay or learning.
| Term | Meaning |
|---|---|
| Recording | One immutable sensor byte buffer. This is the minimum dataset split parent. |
| Trajectory observation | Ordered sensor samples for one training projectile, with time, units, frame, and quality. |
| Region request | An approximate byte interval using [start, end_exclusive). |
| Candidate | One proposed byte interval plus plugin evidence, score, depth, and parent relationships. |
| Case | One region request against one recording. |
| Run | One case evaluated under an exact extractor, policy, plugin, state, adapter, and TDS-version fingerprint. |
| API trace | Ordered public TDS calls, normalized arguments, and stable results. |
| Label | Independently sourced ground truth or review outcome. A selected candidate is a prediction, not a label. |
| Learning sample | Either a complete candidate set or one prefix-conditioned API decision derived from a committed episode. |
- Evaluate immutable
bytes. If capture usesbytearray, convert it once before hashing or evaluation. - All intervals use
[start, end_exclusive). - Candidate exploration uses
memoryviewand does not copy payloads. - The raw recording is stored once. Candidate payloads are reconstructed from the verified source reference plus source hash and offsets; a hash alone cannot reconstruct bytes.
.tobytes()is reserved for the selected payload or an explicit export boundary.- Execution and ingestion wall-clock timestamps are metadata, not identity inputs. Capture timestamps embedded in the source bytes necessarily affect the source hash and its derived case identity.
- Canonical IDs use lowercase SHA-256 digests.
The recovered payload must be parsed into a separate, versioned observation. A string such as x=0,y=0;x=5,y=2 is not sufficient for learning because it has no time basis, units, coordinate frame, or quality.
{
"schema": "staqtapp.sensor_trajectory.v1",
"trajectory_id": "round-1001",
"registry_id": "RID-ABC12345",
"sequence_id": 1001,
"captured_at": "2026-07-22T20:15:00Z",
"sensor_id": "trajectory-array-a",
"coordinate_frame": "sensor_local_enu",
"time_basis": "monotonic_offset_ns",
"units": {
"position": "m"
},
"calibration_fingerprint": "sha256:example",
"samples": [
{
"sample_index": 0,
"t_offset_ns": 0,
"position_m": [0.0, 0.0, 0.0],
"quality_ppm": 990000
},
{
"sample_index": 1,
"t_offset_ns": 10000000,
"position_m": [0.05, 0.02, 0.01],
"quality_ppm": 985000
}
]
}Persist fixed-point values such as quality_ppm and score_ppm in deterministic envelopes. Convert them to floating-point only when a public API requires floats.
The same fixture must be used from extraction through replay. This eliminates the prior mismatch in which the README showed one record but requested and reported two.
import json
from recursive_region_extractor import ByteRegion
def encode_record(observation: dict) -> bytes:
body = json.dumps(
observation,
sort_keys=True,
separators=(",", ":"),
allow_nan=False,
).encode("utf-8")
header = (
f"{observation['registry_id']}|"
f"SEQ-{observation['sequence_id']}|"
f"{observation['captured_at']}|"
).encode("ascii")
return b"<TRAJ>" + header + body + b"</TRAJ>"
observations = (
{
"schema": "staqtapp.sensor_trajectory.v1",
"trajectory_id": "round-1001",
"registry_id": "RID-ABC12345",
"sequence_id": 1001,
"captured_at": "2026-07-22T20:15:00Z",
"sensor_id": "trajectory-array-a",
"coordinate_frame": "sensor_local_enu",
"time_basis": "monotonic_offset_ns",
"units": {"position": "m"},
"calibration_fingerprint": "sha256:calibration-a",
"samples": [
{"sample_index": 0, "t_offset_ns": 0, "position_m": [0.0, 0.0, 0.0], "quality_ppm": 990000},
{"sample_index": 1, "t_offset_ns": 10000000, "position_m": [0.05, 0.02, 0.01], "quality_ppm": 985000},
],
},
{
"schema": "staqtapp.sensor_trajectory.v1",
"trajectory_id": "round-1002",
"registry_id": "RID-ABC12346",
"sequence_id": 1002,
"captured_at": "2026-07-22T20:15:00.020000Z",
"sensor_id": "trajectory-array-a",
"coordinate_frame": "sensor_local_enu",
"time_basis": "monotonic_offset_ns",
"units": {"position": "m"},
"calibration_fingerprint": "sha256:calibration-a",
"samples": [
{"sample_index": 0, "t_offset_ns": 0, "position_m": [0.0, 0.0, 0.0], "quality_ppm": 988000},
{"sample_index": 1, "t_offset_ns": 10000000, "position_m": [0.04, 0.03, 0.01], "quality_ppm": 982000},
],
},
)
buffer = bytearray(b"noise|")
true_regions: dict[str, ByteRegion] = {}
for observation in observations:
encoded = encode_record(observation)
start = len(buffer)
buffer.extend(encoded)
end_exclusive = len(buffer)
true_regions[observation["trajectory_id"]] = ByteRegion(start, end_exclusive)
buffer.extend(b"|noise|")
# Freeze the source before hashing or evaluation.
raw_recording = bytes(buffer)
# Deliberately damage both nominal requests in the same way.
nominal_regions = {
trajectory_id: ByteRegion(region.start + 2, region.end - 3)
for trajectory_id, region in true_regions.items()
}The true_regions mapping is valid ground truth because this fixture is produced by a synthetic generator. In real captures, selected regions remain unverified predictions until an independent review source supplies a label.
Boundary recovery is not trajectory normalization. After selection, parse the envelope, cross-check duplicated identity fields, and validate sample order before storing an AI-ready observation.
import math
def decode_selected_record(payload: bytes) -> dict:
prefix = b"<TRAJ>"
suffix = b"</TRAJ>"
if not payload.startswith(prefix) or not payload.endswith(suffix):
raise ValueError("trajectory delimiters are incomplete")
inner = payload[len(prefix):-len(suffix)]
registry_raw, sequence_raw, captured_raw, body = inner.split(b"|", 3)
observation = json.loads(body.decode("utf-8"))
registry_id = registry_raw.decode("ascii")
sequence_token = sequence_raw.decode("ascii")
captured_at = captured_raw.decode("ascii")
if sequence_token != f"SEQ-{observation['sequence_id']}":
raise ValueError("sequence ID differs between envelope and body")
if registry_id != observation["registry_id"]:
raise ValueError("registry ID differs between envelope and body")
if captured_at != observation["captured_at"]:
raise ValueError("capture time differs between envelope and body")
if observation["schema"] != "staqtapp.sensor_trajectory.v1":
raise ValueError("unsupported trajectory schema")
if observation["coordinate_frame"] != "sensor_local_enu":
raise ValueError("unexpected coordinate frame")
if observation["units"] != {"position": "m"}:
raise ValueError("unexpected position units")
previous_t = -1
for expected_index, sample in enumerate(observation["samples"]):
if sample["sample_index"] != expected_index:
raise ValueError("sample indexes are not contiguous")
if sample["t_offset_ns"] <= previous_t:
raise ValueError("sample times are not strictly increasing")
previous_t = sample["t_offset_ns"]
position = sample["position_m"]
if len(position) != 3 or not all(
isinstance(value, (int, float))
and not isinstance(value, bool)
and math.isfinite(value)
for value in position
):
raise ValueError("position_m must contain three finite numbers")
quality = sample["quality_ppm"]
if not isinstance(quality, int) or not 0 <= quality <= 1_000_000:
raise ValueError("quality_ppm is outside [0, 1000000]")
return observationrecursive_region_extractor.py remains content-neutral. It owns candidate generation and selection, not TDS storage.
Required properties:
- zero-copy
memoryviewslices; - deduplication of identical
[start, end_exclusive)proposals before evaluation; - bounded depth, candidates per depth, and candidates per run;
- content-derived candidate IDs rather than
candidate_0,candidate_1, and similar order-sensitive names; - complete parent and proposal relationships;
- per-plugin verdict, score, rationale code, and proposed correction;
- explicit abstention when no candidate reaches the acceptance policy.
source = memoryview(raw_recording)
candidate = source[start:end_exclusive] # zero-copy
selected_payload = candidate.tobytes() # intentional persistence copyThe baseline policy remains bounded:
| Parameter | Demonstration value | Meaning |
|---|---|---|
acceptance_score |
0.70 |
Minimum aggregate score for resolution. |
uncertainty_floor |
0.20 |
Lowest score allowed to propose further bounded work. |
maximum_depth |
3 |
Maximum recursive correction depth. |
maximum_candidates_per_depth |
256 |
Per-depth work cap. |
maximum_candidates_total |
768 |
Required run-level cap, enforced by the adapter if the extractor lacks it. |
initial_left_window |
8 |
Initial left-boundary search window in bytes. |
initial_right_window |
8 |
Initial right-boundary search window in bytes. |
maximum_initial_length_change |
12 |
Initial region-length variation cap. |
Stateful evidence needs an additional rule. A duplicate index must be frozen, fingerprinted, and built independently inside each dataset split. An external/AI adapter must record its model, configuration, input, and output fingerprints. Otherwise the run is a controlled re-evaluation, not exact replay.
Store one bounded envelope for each case/run. Candidate bytes are not duplicated; every region refers back to the immutable source. The readable digest tokens in this JSON are schematic placeholders; production writers must require sha256: followed by exactly 64 lowercase hexadecimal characters.
{
"schema": "staqtapp.trajectory_episode",
"schema_version": 1,
"case_id": "sha256:case",
"run_id": "sha256:run",
"source": {
"tds_ref": "/trajectory_evidence/recordings/demo-mission/trajectory-array-a/recording-001/source/raw_recording",
"sha256": "sha256:source",
"byte_length": 1194
},
"request": {
"trajectory_id": "round-1001",
"channel_id": "channel-03",
"sequence_id": 1001,
"nominal_region": {"start": 8, "end_exclusive": 583}
},
"execution": {
"tds_release": "3.5.2",
"extractor_build": "sha256:extractor",
"policy_sha256": "sha256:policy",
"plugin_set_sha256": "sha256:plugins",
"plugin_state_sha256": "sha256:state",
"adapter_sha256": "sha256:adapter",
"normalizer_sha256": "sha256:normalizer",
"budget": {
"maximum_depth": 3,
"maximum_candidates_per_depth": 256,
"maximum_candidates_total": 768
},
"usage": {
"depth_reached": 2,
"candidates_evaluated": 413,
"candidates_deduplicated": 71,
"candidates_dropped": 0
}
},
"candidates": [
{
"candidate_id": "sha256:candidate",
"region": {"start": 6, "end_exclusive": 586},
"depth": 1,
"parent_ids": ["sha256:parent"],
"evidence": [
{
"plugin": "delimiter",
"plugin_version": "1",
"verdict": "valid",
"score_ppm": 1000000,
"reason_code": "both_delimiters_present"
}
],
"score_ppm": 901887,
"confidence_ppm": 940000,
"rank": 1
}
],
"decision": {
"status": "resolved",
"selected_candidate_id": "sha256:candidate",
"selected_region": {"start": 6, "end_exclusive": 586},
"selected_payload_sha256": "sha256:selected",
"reason_code": "highest_accepted_rank"
},
"normalized_observation": {
"schema": "staqtapp.sensor_trajectory.v1",
"tds_ref": "/trajectory_evidence/recordings/demo-mission/trajectory-array-a/recording-001/trajectories/round-1001/cases/case-example/runs/run-example/normalized_observation",
"sha256": "sha256:observation"
},
"label": {
"status": "verified",
"source": "synthetic_generator",
"label_version": "1",
"true_region": {"start": 6, "end_exclusive": 586},
"correct_candidate_id": "sha256:candidate"
},
"outcome": {
"boundary_correct": true,
"api_plan_approved": true,
"commit_status": "pending"
}
}Canonicalize objects before hashing. Sort candidate and evidence collections, reject NaN and Infinity, exclude latency and mount-specific paths from identities, and never include execution or ingestion wall-clock timestamps in case_id, run_id, or candidate_id. The json.dumps(..., sort_keys=True) helper below is deterministic for this constrained ASCII/integer demonstration schema; cross-language production identities should use a documented canonical JSON standard such as RFC 8785 rather than assuming sort_keys is a complete canonicalizer.
Recommended identities:
case_id: source hash plus canonical region request;run_id: case ID plus extractor, policy, plugin, state, adapter, normalizer, and TDS-release fingerprints;candidate_id: case ID plus start and end-exclusive offsets.
Hash fields may use the sha256:<hex> notation. TDS directory components should use filesystem-safe forms such as case-<hex>, run-<hex>, and candidate-<hex>.
The adapter API and the TDS API are different contracts.
| Workflow operation | Owner | Call | Required handling |
|---|---|---|---|
| Evaluate a region | Extractor | RecursiveRegionEvaluator.evaluate() |
Preserve candidate graph and abstention. |
| Process a recording | Adapter | WiffleTrajectoryTDSIntegration.process_recording() |
Custom method; do not present it as a TDS call. |
| Store JSON-safe evidence | TDS | TDSDirectory.write_json() |
Check its returned TDSResult, then compare a confirmation read with the intended canonical hash. |
| Store raw or selected bytes | TDS | TDSDirectory.write_result() |
Check TDSResult.ok; it has no overwrite= argument. |
| Rank candidates | TDS Spiral | rank_trace_result() |
Check TDSResult.ok before reading .value. |
| Verify health | TDS | verify(fs) |
Require health.status == "healthy". |
| Persist the tree | TDS | TDSPersistence(...).flush() |
Retain its returned mapping; replacement is atomic per node, not across the whole tree. |
| Read one persisted node | TDS | TDSReader.read_many_result() |
Use fully qualified keys from keys() and close in finally. |
| Replay a trajectory | Adapter | replay_trajectory() |
Custom domain behavior; TDS stores its evidence. |
| Observe diagnostics | TDS | `native_diag_emit(DiagnosticEvent | int, ...)` |
The stable public imports for this guide are:
from pathlib import Path
from staqtapp_tds import TDSFileSystem, TDSPersistence, TDSReader
from staqtapp_tds.diagnostics import (
native_diag_emit,
native_diag_snapshot,
native_diagnostics_available,
)
from staqtapp_tds.spiral.rank import rank_trace_result
from staqtapp_tds.verify import verifyThe extractor must supply an unranked bounded candidate_graph. The adapter calls TDS Spiral first, applies the acceptance policy to that ranking, then constructs the selected payload, normalized observation, episode, and API trace. Selection is never finalized before ranking.
This reference sequence is intentionally restricted to a new mount directory. A production adapter that opens an existing store must use TDSPersistence.mount() or load_node(), compare persisted canonical hashes, return an existing verified commit for an exact match, reject identity conflicts, and create a new run for changed configuration. It must not flush a fresh in-memory tree over an unreconciled store.
Inputs to this fragment are immutable raw_recording and JSON-safe source_manifest, request_manifest, and unranked candidate_graph objects. The request manifest contains no label or later outcome. The graph must include execution, policy.acceptance_score_ppm, and bounded candidates with candidate_id, region, score_ppm, confidence_ppm, and depth. The sequence derives case_id and run_id rather than accepting caller-chosen path names.
from copy import deepcopy
from hashlib import sha256
import json
from pathlib import Path
from staqtapp_tds import TDSFileSystem, TDSPersistence, TDSReader
from staqtapp_tds.spiral.rank import rank_trace_result
from staqtapp_tds.verify import verify
def canonical_json_bytes(value) -> bytes:
return json.dumps(
value,
sort_keys=True,
separators=(",", ":"),
allow_nan=False,
).encode("utf-8")
def artifact_sha256(value) -> str:
if isinstance(value, memoryview):
payload = value.tobytes()
elif isinstance(value, (bytes, bytearray)):
payload = bytes(value)
else:
payload = canonical_json_bytes(value)
return "sha256:" + sha256(payload).hexdigest()
def require_ok(result, operation: str):
if not result.ok:
raise RuntimeError(
f"{operation} failed: {result.code}: {result.message} "
f"path={result.path} meta={result.meta}"
)
return result.value
def stable_result(result) -> dict:
return {
"ok": bool(result.ok),
"code": str(result.code),
"name": result.name,
"path": result.path,
}
def record_result(api_trace, *, phase, api, arguments, result) -> None:
if api_trace is None:
return
api_trace.append(
{
"ordinal": len(api_trace),
"phase": phase,
"api": api,
"arguments": arguments,
"result": stable_result(result),
}
)
def write_json_verified(
directory,
name: str,
value,
*,
provenance: str,
phase: str,
api_trace,
) -> None:
write_result = directory.write_json(
name,
value,
overwrite=False,
provenance=provenance,
)
record_result(
api_trace,
phase=phase,
api="TDSDirectory.write_json",
arguments={
"directory_path": directory.path(),
"name": name,
"value_sha256": artifact_sha256(value),
"overwrite": False,
"provenance": provenance,
},
result=write_result,
)
require_ok(write_result, f"write {name}")
read_result = directory.read_result(name)
record_result(
api_trace,
phase=f"{phase}_confirmation",
api="TDSDirectory.read_result",
arguments={"directory_path": directory.path(), "name": name},
result=read_result,
)
stored = require_ok(read_result, f"confirm {name}")
if artifact_sha256(stored) != artifact_sha256(value):
raise RuntimeError(f"confirmation hash mismatch for {name}")
def portable_hash_component(prefix: str, identity: str) -> str:
marker = "sha256:"
if not identity.startswith(marker):
raise ValueError(f"{prefix} identity must start with {marker}")
digest = identity[len(marker):]
if len(digest) != 64 or any(ch not in "0123456789abcdef" for ch in digest):
raise ValueError(f"{prefix} identity is not lowercase SHA-256")
return f"{prefix}-{digest}"
def portable_name_component(value: str) -> str:
allowed = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789._-"
if not value or value in {".", ".."} or any(ch not in allowed for ch in value):
raise ValueError(f"unsafe TDS path component: {value!r}")
return value
mount_dir = Path("./trajectory_tds_store_new")
if mount_dir.exists():
raise FileExistsError(
"This new-store example refuses an existing mount. "
"Mount/load and reconcile it before writing."
)
if artifact_sha256(raw_recording) != source_manifest["sha256"]:
raise ValueError("source manifest hash does not match raw_recording")
case_id = artifact_sha256(
{
"source_sha256": source_manifest["sha256"],
"request": request_manifest,
}
)
run_id = artifact_sha256(
{
"case_id": case_id,
"execution": candidate_graph["execution"],
"policy": candidate_graph["policy"],
}
)
for candidate in candidate_graph["candidates"]:
expected_candidate_id = artifact_sha256(
{"case_id": case_id, "region": candidate["region"]}
)
if candidate["candidate_id"] != expected_candidate_id:
raise ValueError("candidate ID does not match case ID and region")
case_component = portable_hash_component("case", case_id)
run_component = portable_hash_component("run", run_id)
fs = TDSFileSystem("trajectory_evidence")
store = TDSPersistence(mount_dir)
api_trace: list[dict] = []
mission_component = portable_name_component(source_manifest["mission_id"])
sensor_component = portable_name_component(source_manifest["sensor_id"])
recording_component = portable_name_component(source_manifest["recording_id"])
trajectory_component = portable_name_component(request_manifest["trajectory_id"])
base = (
f"/recordings/{mission_component}/{sensor_component}/{recording_component}"
)
case_path = (
f"{base}/trajectories/{trajectory_component}/cases/{case_component}"
)
run_path = f"{case_path}/runs/{run_component}"
source_dir = fs.makedirs(f"{base}/source")
case_dir = fs.makedirs(case_path)
run_dir = fs.makedirs(run_path)
source_write = source_dir.write_result(
"raw_recording",
raw_recording,
provenance="immutable-sensor-source",
)
record_result(
api_trace,
phase="persist_source",
api="TDSDirectory.write_result",
arguments={
"directory_path": source_dir.path(),
"name": "raw_recording",
"value_sha256": artifact_sha256(raw_recording),
"provenance": "immutable-sensor-source",
},
result=source_write,
)
require_ok(source_write, "write raw recording")
write_json_verified(
source_dir,
"manifest",
source_manifest,
provenance="recording-source-manifest-v1",
phase="persist_source_manifest",
api_trace=api_trace,
)
write_json_verified(
case_dir,
"request_manifest",
request_manifest,
provenance="trajectory-region-request-v1",
phase="persist_request",
api_trace=api_trace,
)
# TDS Spiral ranks the unranked bounded graph before any selection is built.
candidate_ids = [item["candidate_id"] for item in candidate_graph["candidates"]]
scores = [item["score_ppm"] / 1_000_000 for item in candidate_graph["candidates"]]
confidences = [
item["confidence_ppm"] / 1_000_000
for item in candidate_graph["candidates"]
]
depths = [item["depth"] for item in candidate_graph["candidates"]]
rank_result = rank_trace_result(
candidate_ids,
scores,
confidences=confidences,
depths=depths,
)
record_result(
api_trace,
phase="rank_candidates",
api="rank_trace_result",
arguments={
"trace_ids": candidate_ids,
"scores": scores,
"confidences": confidences,
"depths": depths,
"normalization": "score_ppm and confidence_ppm divided by 1000000",
},
result=rank_result,
)
rank_records = require_ok(rank_result, "rank candidates")
ranking = [
{
"rank": item.rank,
"candidate_id": item.trace_id,
"rank_score_ppm": int(round(item.rank_score * 1_000_000)),
}
for item in rank_records
]
rank_by_id = {item["candidate_id"]: item for item in ranking}
ranked_graph = deepcopy(candidate_graph)
for candidate in ranked_graph["candidates"]:
ranked = rank_by_id[candidate["candidate_id"]]
candidate["rank"] = ranked["rank"]
candidate["rank_score_ppm"] = ranked["rank_score_ppm"]
ranked_graph["candidates"].sort(key=lambda item: item["rank"])
acceptance_score_ppm = ranked_graph["policy"]["acceptance_score_ppm"]
selected = next(
(
item
for item in ranked_graph["candidates"]
if item["score_ppm"] >= acceptance_score_ppm
),
None,
)
selected_payload = None
normalized_observation = None
if selected is None:
selected_summary = {
"status": "unresolved",
"selected_candidate_id": None,
"selected_region": None,
"selected_payload_sha256": None,
"reason_code": "no_candidate_met_acceptance_policy",
}
else:
start = selected["region"]["start"]
end_exclusive = selected["region"]["end_exclusive"]
if not 0 <= start < end_exclusive <= len(raw_recording):
raise ValueError("selected region is outside the immutable source")
selected_payload = memoryview(raw_recording)[start:end_exclusive].tobytes()
normalized_observation = decode_selected_record(selected_payload)
selected_summary = {
"status": "resolved",
"selected_candidate_id": selected["candidate_id"],
"selected_region": {
"start": start,
"end_exclusive": end_exclusive,
},
"selected_payload_sha256": artifact_sha256(selected_payload),
"reason_code": "highest_accepted_spiral_rank",
}
root_key = "/trajectory_evidence"
run_key = f"{root_key}{run_path}"
normalized_ref = (
f"{run_key}/normalized_observation"
if normalized_observation is not None
else None
)
episode = {
"schema": "staqtapp.trajectory_episode",
"schema_version": 1,
"case_id": case_id,
"run_id": run_id,
"source": {
"tds_ref": f"{root_key}{base}/source/raw_recording",
"sha256": source_manifest["sha256"],
"byte_length": len(raw_recording),
},
"request": request_manifest,
"execution": ranked_graph["execution"],
"candidates": ranked_graph["candidates"],
"decision": selected_summary,
"normalized_observation": (
{
"schema": normalized_observation["schema"],
"tds_ref": normalized_ref,
"sha256": artifact_sha256(normalized_observation),
}
if normalized_observation is not None
else None
),
"label": {"status": "unverified"},
"outcome": {"commit_status": "pending"},
}
for name, value, provenance, phase in (
("candidate_graph", ranked_graph, "bounded-extraction-evidence-v1", "persist_candidates"),
("candidate_ranking", ranking, "tds-spiral-ranking-v3.5.2", "persist_ranking"),
("selected_summary", selected_summary, "trajectory-selection-summary-v1", "persist_selection"),
("episode", episode, "trajectory-evidence-episode-v1", "persist_episode"),
):
write_json_verified(
run_dir,
name,
value,
provenance=provenance,
phase=phase,
api_trace=api_trace,
)
if selected_payload is not None:
payload_write = run_dir.write_result(
"selected_payload",
selected_payload,
provenance="selected-derived-source-region",
)
record_result(
api_trace,
phase="persist_selected_payload",
api="TDSDirectory.write_result",
arguments={
"directory_path": run_dir.path(),
"name": "selected_payload",
"value_ref": (
f"source:[{selected_summary['selected_region']['start']},"
f"{selected_summary['selected_region']['end_exclusive']})"
),
"value_sha256": artifact_sha256(selected_payload),
"provenance": "selected-derived-source-region",
},
result=payload_write,
)
require_ok(payload_write, "write selected payload")
write_json_verified(
run_dir,
"normalized_observation",
normalized_observation,
provenance="sensor-trajectory-v1",
phase="persist_normalized_observation",
api_trace=api_trace,
)
# The trace is built from actual TDSResult objects. Its own write/read is excluded
# to prevent recursive self-recording.
write_json_verified(
run_dir,
"api_trace",
api_trace,
provenance="public-tds-call-trace-v1",
phase="persist_api_trace",
api_trace=None,
)
health = verify(fs)
if health.status != "healthy":
raise RuntimeError(f"TDS health gate failed: {health}")
flush_mapping = store.flush(fs, parallel_nodes=True)The stored episode remains pending at this point. verify(fs) proves the in-memory tree is healthy; it does not prove every persisted node can be reopened. The adapter must read back and hash-check every required artifact before publishing a commit.
Every TDS directory is persisted as its own .tds node. The helper below groups fully qualified keys by the node that actually contains them, uses result-first reads, and closes every reader even when keys() or decoding fails.
def read_persisted_values(tds_paths, required_keys):
remaining = set(required_keys)
values = {}
for tds_path in tds_paths:
reader = TDSReader(tds_path)
try:
available = sorted(remaining.intersection(reader.keys()))
if not available:
continue
result = reader.read_many_result(available)
batch = require_ok(result, f"read persisted node {tds_path}")
values.update(batch)
remaining.difference_update(available)
finally:
reader.close()
if remaining:
raise KeyError(f"persisted keys not found: {sorted(remaining)}")
return values
source_key = f"{root_key}{base}/source/raw_recording"
source_manifest_key = f"{root_key}{base}/source/manifest"
request_key = f"{root_key}{case_path}/request_manifest"
expected_by_key = {
source_key: raw_recording,
source_manifest_key: source_manifest,
request_key: request_manifest,
f"{run_key}/candidate_graph": ranked_graph,
f"{run_key}/candidate_ranking": ranking,
f"{run_key}/selected_summary": selected_summary,
f"{run_key}/episode": episode,
f"{run_key}/api_trace": api_trace,
}
if selected_payload is not None:
expected_by_key[f"{run_key}/selected_payload"] = selected_payload
expected_by_key[f"{run_key}/normalized_observation"] = normalized_observation
persisted = read_persisted_values(flush_mapping.keys(), expected_by_key)
artifact_digests = {}
for key, expected in expected_by_key.items():
expected_digest = artifact_sha256(expected)
actual_digest = artifact_sha256(persisted[key])
if actual_digest != expected_digest:
raise RuntimeError(f"persisted artifact hash mismatch: {key}")
artifact_digests[key] = actual_digest
# Adapter-defined commit manifest: created only after complete read-back.
commit_manifest = {
"schema": "staqtapp.trajectory_commit.v1",
"case_id": case_id,
"run_id": run_id,
"health_status": health.status,
"required_artifacts": artifact_digests,
"artifact_node_count": len(flush_mapping),
"readback_verified": True,
"terminal_calls": [
{
"api": "verify",
"arguments": {"target": "trajectory_evidence"},
"result": {"status": health.status},
},
{
"api": "TDSPersistence.flush",
"arguments": {"parallel_nodes": True},
"result": {"node_count": len(flush_mapping)},
},
{
"api": "TDSReader.read_many_result",
"arguments": {"required_key_count": len(expected_by_key)},
"result": {"all_hashes_verified": True},
},
],
}
commit_dir = fs.makedirs(f"{base}/commits")
write_json_verified(
commit_dir,
run_component,
commit_manifest,
provenance="trajectory-run-commit-v1",
phase="persist_commit",
api_trace=None,
)
commit_health = verify(fs)
if commit_health.status != "healthy":
raise RuntimeError(f"commit health gate failed: {commit_health}")
# Persist only the commit directory, serially, so it is the last published node.
commit_tds_path, _ = store.flush_node(commit_dir, parallel=False)
commit_key = f"{root_key}{base}/commits/{run_component}"
commit_reader = TDSReader(commit_tds_path)
try:
committed = require_ok(
commit_reader.read_result(commit_key),
"read back commit manifest",
)
finally:
commit_reader.close()
if artifact_sha256(committed) != artifact_sha256(commit_manifest):
raise RuntimeError("commit manifest hash mismatch")Dataset readers begin from verified commit manifests and reject a run when any named artifact is absent or has a different hash. The commit manifest is an adapter protocol, not a built-in cross-node TDS transaction.
A storage layout alone does not teach an AI how the evidence was produced. Each run therefore stores an ordered API trace constructed from the actual calls and returned TDSResult objects. The following is an abridged schematic excerpt; retained ordinals show that other calls occurred between the displayed entries, and readable digest tokens stand in for validated 64-character hashes.
[
{
"ordinal": 0,
"phase": "persist_source",
"api": "TDSDirectory.write_result",
"arguments": {
"directory_path": "/trajectory_evidence/recordings/demo-mission/trajectory-array-a/recording-001/source",
"name": "raw_recording",
"value_sha256": "sha256:source",
"provenance": "immutable-sensor-source"
},
"result": {
"ok": true,
"code": "OK",
"path": "/trajectory_evidence/recordings/demo-mission/trajectory-array-a/recording-001/source/raw_recording"
}
},
{
"ordinal": 5,
"phase": "rank_candidates",
"api": "rank_trace_result",
"arguments": {
"trace_ids": ["sha256:candidate"],
"scores": [0.901887],
"confidences": [0.94],
"depths": [1],
"normalization": "score_ppm and confidence_ppm divided by 1000000"
},
"result": {
"ok": true,
"code": "OK",
"value_ref": "candidate_ranking"
}
}
]Trace rules:
- Record only public API names.
- Normalize byte or large-value arguments to content references; do not duplicate payloads.
- Persist stable result fields such as
ok,code,name, and logicalpath. - Exclude transient latency, host paths, object addresses, and diagnostic timing from deterministic fingerprints.
- For an API-call learning sample, include only the episode prefix known before the call. Future results are labels or outcomes, not input features.
- The trace excludes the act of storing itself to avoid recursive self-recording. Health, flush, persisted read-back, and final commit outcomes belong to the adapter-defined commit manifest.
/trajectory_evidence/
recordings/
demo-mission/
trajectory-array-a/
recording-001/
source/
manifest JSON
raw_recording bytes
trajectories/
round-1001/
cases/
<content-derived-case-id>/
request_manifest JSON
runs/
<content-derived-run-id>/
episode JSON
candidate_graph JSON
candidate_ranking JSON
selected_summary JSON
selected_payload bytes
normalized_observation JSON
api_trace JSON
review_label JSON, optional
replay/
round-1001/
<content-derived-replay-id> JSON
recording_summary JSON
commits/
<content-derived-run-id> JSON, written last
The raw recording is source evidence. Everything else is derived, versioned, and attributable. Storing one bounded candidate graph avoids one TDS directory per candidate and avoids copying overlapping candidate payloads.
Because a tree flush is not a cross-node transaction, publish a small commit manifest only after the required source and run nodes have been flushed and read back successfully. Flush that commit node last. Dataset readers must ignore any run without a valid commit manifest that names and hashes every required artifact.
Keep these as separate fields:
| Object | Examples |
|---|---|
| Ground-truth boundary | verified region, ambiguous, no valid region, unknown |
| Prediction | selected candidate, abstention |
| Label provenance | synthetic generator, human review, cross-sensor consensus, unverified pseudo-label |
| API-plan label | approved, incorrect call, missing call, unnecessary call, unknown |
| TDS operational result | TDSResult.ok, code, logical path, stable metadata |
| Storage outcome | committed, incomplete, health verification failed |
| Final correctness | correct selection, false selection, correct abstention, false abstention, unknown |
resolved means the policy accepted a candidate. It does not mean the boundary is correct. Unverified selections must never silently become gold labels.
Retain malformed, truncated, ambiguous, conflicting-ID, duplicate, storage-failure, and no-valid-candidate cases so an AI can learn correct abstention and failure handling.
- Split at mission or recording-session level, never at candidate level.
- Keep every candidate, run, replay, label revision, and API step for one case in the same split.
- Group duplicate recordings and observations of the same physical event before assigning splits.
- Build duplicate indexes independently for train, validation, and test.
- Freeze external adapters used to create validation and test traces.
- Use a temporal holdout for replay and drift evaluation.
- Freeze test labels; a label correction creates a new dataset version.
- Fit calibration or normalization statistics on training data only.
- Store a split-manifest fingerprint and duplicate-cluster IDs.
- Export a candidate set as one contrastive example; do not split overlapping candidates into independent rows.
Exact replay requires identical:
- source bytes and case request;
- extractor and normalizer builds;
- plugin order, versions, and parameters;
- duplicate/plugin state snapshot;
- external-adapter model, configuration, input, and output;
- TDS release and Spiral ranking inputs.
Any change to candidate set, scores, ranks, selection, normalized observation, API plan, or API result is a replay regression or nondeterminism signal.
A deliberate policy, plugin, adapter, normalizer, or state change creates a new run under the same case. Compare it as drift; do not label it exact replay.
Report drift on independent axes:
source
configuration
evidence_state
candidate_set
scores_and_ranks
selected_region
selected_payload
normalized_observation
api_plan
api_result
storage_verification
If scores change while selected bytes remain stable, report decision_stable, not unchanged.
Native diagnostics are a bounded observer side channel, not durable evidence.
native_diag_emit()accepts a documentedDiagnosticEventor integer code, not an arbitrary string.- The adapter must version any human-readable name-to-code mapping.
- A full diagnostic ring drops events and increments a counter instead of blocking storage.
Falsefrom a diagnostic emit means the emit was not accepted. Inspect availability, status, and snapshot evidence for the reason; do not infer that the observed event did not occur.- If a diagnostic snapshot becomes model input, reference the exact immutable snapshot observed before the action.
The previous README reported test-double observations, which do not prove compatibility with the real TDS v3.5.2 implementation. A publishable validation record must execute the actual extractor, adapter, and TDS package and verify:
- both records in the canonical fixture are recovered;
ExtractionDumpRegistryand every example import resolve;- source, candidate, ranking, selection, observation, and API-trace artifacts persist;
- every expected
TDSResultis checked; - Python and native Spiral ranking agree;
- candidate and run fingerprints are order-independent;
- the global candidate cap is enforced;
- duplicate state is split-local and replayable;
- health verification gates the flush;
- persisted paths come from the flush mapping;
- fully qualified reader keys match
TDSReader.keys(); - readers close on success and failure;
- exact replay and controlled re-evaluation are classified separately;
- partial runs without a final commit marker are ignored by dataset readers.
Until those tests run against the two Python modules and TDS v3.5.2, describe the document as API-reviewed, not runtime-validated.
recursive_region_extractor
owns bounded byte-region evaluation and candidate evidence
trajectory plugins
own domain-specific evidence and proposals
wiffle_trajectory_tds_integration
owns normalization, episode assembly, replay classification,
and translation into public TDS calls
Staqtapp-TDS
owns storage, provenance, Spiral ranking, health verification,
per-node atomic persistence, diagnostics, and validated reads
external trainer or evaluator
owns dataset selection, labels, splits, training, and metrics
Create one Gist or repository example containing:
README.md
recursive_region_extractor.py
wiffle_trajectory_tds_integration.py
demo_recording.py
test_recursive_trajectory_tds.py
The demo must build the canonical two-record fixture. The test must run against the actual supported Staqtapp-TDS release rather than only a compatible test double.
Preserve the source. Explain the selection. Record the call. Verify the outcome.
Higher educational matters should take note that TDS is being used with spline-shift data feedback. Staqtapp-1Xq is not TDS, 1Xq is the flagship Staqtapp storage intelligence model for standard NN.