Compare the security terminal observation at second granularity

src/github_repo_auditor/portfolio_generation.py

@@ -106,6 +106,17 @@ def _parse_instant(value: object, *, field: str) -> datetime:
     return parsed.astimezone(UTC)
 
 
+def _terminal_observed_in_order(
+    terminal_observed: datetime, produced_at: datetime, now: datetime
+) -> bool:
+    """A terminal receipt may not be observed before the receipt it attests, nor
+    in the future. Terminal receipts carry a whole-second observed_at while
+    security receipts carry microseconds, so a terminal written in the same
+    second as the receipt (the normal case for a fast collector) compares at
+    second granularity; anything a full second earlier is still refused."""
+    return produced_at.replace(microsecond=0) <= terminal_observed <= now.astimezone(UTC)
+
+
 def _instant_text(value: datetime) -> str:
     return value.astimezone(UTC).isoformat().replace("+00:00", "Z")
 
@@ -322,7 +333,7 @@ def load_security_binding(
     terminal_observed = _parse_instant(
         terminal.get("observed_at"), field="security terminal observed_at"
     )
-    if terminal_observed < produced_at or terminal_observed > now.astimezone(UTC):
+    if not _terminal_observed_in_order(terminal_observed, produced_at, now):
         raise PortfolioGenerationError("security terminal observation time is invalid")
     evidence = destination.get("evidence")
     if not isinstance(evidence, dict):

tests/test_portfolio_generation.py

@@ -8,6 +8,7 @@
 from pathlib import Path
 
 from github_repo_auditor.portfolio_generation import (
+    _terminal_observed_in_order,
     ArtifactInput,
     PortfolioGenerationError,
     ProducerBinding,
@@ -275,3 +276,19 @@ def test_rollback_and_roll_forward_swap_the_exact_pair(self):
 
 if __name__ == "__main__":
     unittest.main()
+
+
+class TerminalObservationOrderTests(unittest.TestCase):
+    def test_same_second_terminal_is_accepted_and_earlier_second_is_refused(self) -> None:
+        from datetime import UTC, datetime
+
+        produced = datetime(2026, 9, 2, 12, 8, 8, 576236, tzinfo=UTC)
+        now = datetime(2026, 9, 2, 12, 10, 0, tzinfo=UTC)
+        same_second = datetime(2026, 9, 2, 12, 8, 8, tzinfo=UTC)
+        earlier = datetime(2026, 9, 2, 12, 8, 7, tzinfo=UTC)
+        later = datetime(2026, 9, 2, 12, 8, 9, tzinfo=UTC)
+        future = datetime(2026, 9, 2, 12, 11, 0, tzinfo=UTC)
+        self.assertTrue(_terminal_observed_in_order(same_second, produced, now))
+        self.assertTrue(_terminal_observed_in_order(later, produced, now))
+        self.assertFalse(_terminal_observed_in_order(earlier, produced, now))
+        self.assertFalse(_terminal_observed_in_order(future, produced, now))