fix(agt_data,data-flow): reject float classification tiers and pre-epoch evaluated_at timestamps in Python and TypeScript (#112)

packages/typescript/src/data-flow.ts

@@ -34,14 +34,15 @@ export function agtDataClassification(source: Record<string, unknown>): Classifi
 }
 
 export function agtDataAccessFlow(factory: EventFactory, decision: Record<string, unknown>, options: {runId: string; direction: string; source: DataEndpoint; destination: DataEndpoint; purpose: string; contentDigest?: Digest; sizeBytes?: number; tokenCount?: number; mediaType?: string; transformation?: string; envelope?: Omit<EnvelopeFields, "runId" | "agentId" | "timeUnixNano">}): NormalizedEvent {
-  if (typeof decision.allowed !== "boolean") throw new Error("AGT data access allowed must be a boolean");
-  const label = object(decision.data_label, "AGT data_label");
+  const rec = object(decision, "AGT decision");
+  if (typeof rec.allowed !== "boolean") throw new Error("AGT data access allowed must be a boolean");
+  const label = object(rec.data_label, "AGT data_label");
   const classification = agtDataClassification(label);
-  return classifiedDataFlow(factory, {classify: () => classification}, undefined, {...options, agentId: required(decision.agent_id, "AGT data access agent_id"), policyDecision: decision.allowed ? "allow" : "deny", envelope: {timeUnixNano: timestampNs(decision.evaluated_at, "evaluated_at"), ...options.envelope}});
+  return classifiedDataFlow(factory, {classify: () => classification}, undefined, {...options, agentId: required(rec.agent_id, "AGT data access agent_id"), policyDecision: rec.allowed ? "allow" : "deny", envelope: {timeUnixNano: timestampNs(rec.evaluated_at, "evaluated_at"), ...options.envelope}});
 }
 
 function identifier(value: unknown, field: string): string { if (typeof value !== "string" || !IDENTIFIER.test(value) || value.includes("://")) throw new Error(`${field} must be a 1-256 character metadata identifier without whitespace, URLs, query strings, fragments, or key/value delimiters`); return value; }
 function required(value: unknown, field: string): string { if (typeof value !== "string" || !value) throw new Error(`${field} must be a non-empty string`); return value; }
 function object(value: unknown, field: string): Record<string, unknown> { if (typeof value !== "object" || value === null || Array.isArray(value)) throw new Error(`${field} must be an object`); return value as Record<string, unknown>; }
 function present(key: string, value: unknown): Record<string, unknown> { return value === undefined ? {} : {[key]: value}; }
-function timestampNs(value: unknown, field: string): bigint { if (typeof value !== "string") throw new Error(`AGT data access ${field} must be an RFC 3339 UTC timestamp`); const match = /^(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2})(?:\.(\d{1,9}))?(?:Z|\+00:00)$/.exec(value); if (!match) throw new Error(`AGT data access ${field} must be an RFC 3339 UTC timestamp`); const milliseconds = Date.parse(`${match[1]}Z`); if (match[1]!.startsWith("0000") || !Number.isFinite(milliseconds) || new Date(milliseconds).toISOString().slice(0, 19) !== match[1]) throw new Error(`AGT data access ${field} is invalid`); return BigInt(milliseconds) * 1_000_000n + BigInt((match[2] ?? "").padEnd(9, "0") || "0"); }
+function timestampNs(value: unknown, field: string): bigint { if (typeof value !== "string") throw new Error(`AGT data access ${field} must be an RFC 3339 UTC timestamp`); const match = /^(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2})(?:\.(\d{1,9}))?(?:Z|\+00:00)$/.exec(value); if (!match) throw new Error(`AGT data access ${field} must be an RFC 3339 UTC timestamp`); const milliseconds = Date.parse(`${match[1]}Z`); if (match[1]!.startsWith("0000") || !Number.isFinite(milliseconds) || milliseconds < 0 || new Date(milliseconds).toISOString().slice(0, 19) !== match[1]) throw new Error(`AGT data access ${field} is invalid`); return BigInt(milliseconds) * 1_000_000n + BigInt((match[2] ?? "").padEnd(9, "0") || "0"); }

packages/typescript/test/data-flow.test.ts

@@ -19,4 +19,6 @@ test("AGT access maps only tier, decision, agent, and time", () => {
   const event = agtDataAccessFlow(factory, decision, {runId: "run-1", direction: "read", source: new DataEndpoint("database", "customer-records"), destination: new DataEndpoint("agent", "builder"), purpose: "requirements-analysis"});
   assert.equal((event.classification as Record<string, unknown>).value, "confidential"); assert.equal(event.policy_decision, "deny");
   const serialized = JSON.stringify(event); for (const value of ["PII", "customer@example", "alice@example", "123-45-6789", "geography", "matched_policy"]) assert.equal(serialized.includes(value), false);
+  assert.throws(() => agtDataAccessFlow(factory, {...decision, evaluated_at: "1969-12-31T23:59:59Z"}, {runId: "run-1", direction: "read", source: new DataEndpoint("database", "customer-records"), destination: new DataEndpoint("agent", "builder"), purpose: "requirements-analysis"}), /evaluated_at is invalid/);
+  assert.throws(() => agtDataAccessFlow(factory, null as unknown as Record<string, unknown>, {runId: "run-1", direction: "read", source: new DataEndpoint("database", "customer-records"), destination: new DataEndpoint("agent", "builder"), purpose: "requirements-analysis"}), /must be an object/);
 });

src/agentrust_telemetry/adapters/agt_data.py

@@ -26,7 +26,7 @@ def agt_data_classification(source: Any) -> ClassificationResult:
     """Map only AGT's ordered sensitivity tier; omit auxiliary label content."""
     classification = _field(source, "classification")
     raw = classification.value if isinstance(classification, Enum) else classification
-    if isinstance(raw, bool) or not isinstance(raw, (int, float)) or raw not in _LEVELS:
+    if type(raw) is not int or raw not in _LEVELS:
         raise ValueError(f"unsupported AGT data classification: {raw!r}")
     return ClassificationResult(
         taxonomy="agt.data_classification.v1",
@@ -93,10 +93,12 @@ def _field(source: Any, name: str) -> Any:
 def _datetime_ns(value: Any) -> int:
     from datetime import datetime, timezone
 
-    if not isinstance(value, datetime) or value.tzinfo is None:
+    if not isinstance(value, datetime) or value.tzinfo is None or value.utcoffset() is None:
         raise ValueError("AGT data access evaluated_at must be a timezone-aware datetime")
     utc = value.astimezone(timezone.utc)
     epoch = datetime(1970, 1, 1, tzinfo=timezone.utc)
+    if utc < epoch:
+        raise ValueError("AGT data access evaluated_at must not predate the Unix epoch")
     delta = utc - epoch
     return (
         delta.days * 86_400_000_000_000

tests/test_agt_adapter.py

@@ -155,6 +155,34 @@ def test_sink_uses_source_result_sentinels_and_reports_projection_failure(self):
         )
         self.assertEqual(sink.emit([FakeAgtEvent()]), Result.FAILURE)
 
+    def test_agt_data_rejects_float_tiers_and_pre_epoch_timestamps(self):
+        from datetime import datetime, timezone
+        from agentrust_telemetry.adapters.agt_data import (
+            DataEndpoint,
+            agt_data_access_flow,
+            agt_data_classification,
+        )
+
+        for bad_tier in (0.0, 2.0, True, False, -1, 5, "public"):
+            with self.subTest(bad_tier=bad_tier):
+                with self.assertRaises(ValueError):
+                    agt_data_classification({"classification": bad_tier})
+        with self.assertRaises(ValueError):
+            agt_data_access_flow(
+                self.factory,
+                {
+                    "allowed": True,
+                    "agent_id": "agent-1",
+                    "data_label": {"classification": 1},
+                    "evaluated_at": datetime(1969, 12, 31, 23, 59, 59, tzinfo=timezone.utc),
+                },
+                run_id="run-1",
+                direction="read",
+                source=DataEndpoint("database", "db"),
+                destination=DataEndpoint("agent", "a1"),
+                purpose="analysis",
+            )
+
 
 if __name__ == "__main__":
     unittest.main()