diff --git a/CHANGELOG.md b/CHANGELOG.md index 4a527c2..1d8aa1c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,139 @@ and this project adheres to [Semantic Versioning 2.0.0](https://semver.org/spec/ ## [Unreleased] +### Changed + +- **DDS pure logic extracted for testability (Lot 0, external audit + 2026-07-08).** The QoS-profile normalizers (`_cyclone_qos_to_profile` / + `_fast_qos_to_profile`) and the discovery-sample field extractors + (`_extract_guid` / `_extract_vendor_id` / `_extract_hostname` / + `_extract_topic_name` / `_is_removal`) were moved out of + `adapters/dds_cyclone/adapter.py` and `adapters/dds_fast/adapter.py` — + which import their vendor binding at module top level and were therefore + never exercised by the test suite — into the binding-free + `adapters/common/qos_normalize.py` and `adapters/common/dds_introspection.py`. + The adapters import them back under their original private names, so every + call site is unchanged (verified by `ruff check` static analysis, since the + adapters are not importable without their SDKs). `fast_qos_to_profile` + takes the binding's int→str enum maps as parameters so it stays + import-free. This is the highest-value item from the audit: the QoS + normalization feeds `detect_qos_mismatches` (the flagship DDS diagnostic) + and was previously untestable and untested. + +### Added + +- `tests/test_dds_qos_normalization.py` and `tests/test_dds_introspection.py` + drive the extracted logic with synthetic duck-typed objects (no + `cyclonedds` / `fastdds` needed), including regression guards for the + "renamed policy key silently yields no QoS profile → no mismatch ever + reported" failure mode. The extracted modules are now ~91–92% covered. +- `pytest-cov` and `rosbags` added to the `[dev]` extra, plus `[tool.coverage]` + config with a `fail_under = 85` floor (the two binding-only adapter shells + are `omit`ted as structurally unreachable without their SDKs). Adding + `rosbags` un-skips the real `.db3` bag-analysis I/O test. + +### Fixed + +- `tests/test_bag_service.py` bag-generation helper updated for the current + `rosbags` API (`Writer(..., version=Writer.VERSION_LATEST)`; typestore keyed + by `std_msgs/msg/String`, not `std_msgs__msg__String`). The test was + previously auto-skipped and had gone silently stale — un-skipping it in CI + surfaced the drift. +- **`topic_metrics` frequency was wrong (Lot 2, audit C5).** It divided the + sample count by `(now − oldest_sample)` — folding in idle time since the + last peek — and counted N intervals instead of N−1. Now measured as + `(N−1) / (newest − oldest)` over the samples' own arrival span; samples + surfaced by a single opportunistic peek share one timestamp (span 0) and + correctly yield `frequency_hz_observed = null` instead of a fabricated rate. +- **`topic_metrics` sequence-gap count exploded on multi-writer topics, + publisher restarts, and counter wrap (Lot 2, audit C6).** Gaps are now + counted per writer (new best-effort `MetricsSample.writer_guid`), so + independent writers' counter offsets are not read as phantom gaps, and a + single jump wider than 10 000 is treated as a reset/wrap discontinuity + rather than that many losses. +- **QoS Deadline false negative (Lot 2, audit C3/P1-3).** + `detect_qos_mismatches` now flags a reader that requests a finite Deadline + against a writer that offers none — an absent deadline is the infinite + (loosest) period and cannot satisfy a finite request. The previous rule + required both sides non-null and silently missed this incompatibility. +- **`LifecycleBuffer` participant map was unbounded (Lot 3, audit P1-4 / M1 / + P1).** Only the event ring was capped; the participant dict grew one entry + per GUID ever seen (a churny bus mints a fresh RTPS GUID on each node + restart), and `list_participants` returned every tombstone forever. Now + capped at `MAX_PARTICIPANTS = 4096`, evicting `"left"` tombstones first then + the oldest-inserted entry — the docstring's "Bounded" claim is now true. +- **`MetricsBuffer` topic map was unbounded (Lot 3, audit P2-5).** Per-topic + rings were capped but the number of topic keys was not; now capped at + `MAX_TOPICS = 4096`, oldest-inserted topic evicted on overflow. +- **OpenDDS stub `is_available()` always returns False (Lot 4, audit S1).** A + stub that advertised availability (when a `pyopendds` module happened to be + importable) could be auto-selected by the factory, after which every tool + call raised. Now consistent with the Dust stub. +- **`iter_field_names` mis-decoded a string `__slots__` (Lot 4, audit C2).** + `__slots__ = "value"` was exploded into `['v','a','l','u','e']`; a bare + string slot is now treated as a single field name. +- **`decode_field_value` recursion is depth-capped at 32 (Lot 4, audit M6).** + A pathologically deep decoded object graph collapses to `repr()` instead of + risking `RecursionError`. +- **`_encode_raw_bytes` slices before hex-encoding (Lot 4, audit M5).** A large + raw payload no longer allocates its full 2×-size hex string only to truncate + it to the 4096-char preview. + +### Added (Lot 4 — test hardening) + +- End-to-end test that a failing tool call surfaces as an MCP error + (`ToolError`) rather than being masked as a success — pins the thin-handler + contract (CLAUDE.md §8, audit test-gap #4). +- Test pinning that every canonical vendor tag is a valid + `ParticipantInfo`/`ParticipantEvent.vendor` Literal (guards against + vendor-map ↔ schema drift, audit P2-3). Scenario allowlist `_KNOWN_TOOLS` + now includes the 11th tool `peek_bag_samples`. + +### Removed (Lot 4) + +- Dead `annotate_full` / `annotate_partial` imports and the `_ = (...)` + unused-suppressor from `services/bag_service.py`. + +### Documentation (Lot 1 — reconcile the strategic source of truth) + +- **`docs/product-plan.md` realigned on the shipped 11-tool surface (audit + M6/P1-6).** §1 and §4 said "five typed tools today" / DDS "roadmapped" + while six DDS/observability tools had shipped across v0.2.0–v0.4.0. §11's + risk register carried a self-imposed governance gate — "any 9th tool needs + an explicit re-scope discussion documented in this register before code + lands" — that was crossed during v0.4.0 without the discussion being + recorded. Added a retroactive re-scope decision closing that gap: the three + ceiling-breaking tools are accepted, the new ceiling is 11 tools, a 12th + needs a documented re-scope. +- **User-topic `raw` decode honesty (audit C1).** README and the + `peek_dds_samples` tool description no longer imply the `raw` fallback + preserves the payload in `_raw_bytes_hex` — on the current user-topic raw + path that field is empty (a `raw` status means "present but not decoded"). + Capturing the on-wire CDR bytes is stated as roadmapped rather than done. + +### Changed (Lot 5 — DDS adapter deduplication) + +- **QoS-mismatch endpoint pairing deduplicated (audit D1/M7/P2).** The ~40 + identical lines in each adapter's `detect_qos_mismatches` (group endpoints by + topic, pair reader × writer, build `MismatchReport`) moved to the binding-free + `common/qos_endpoints.detect_mismatches_across_endpoints`, unit-tested without + a binding. Each writer's QoS profile is now parsed once per topic instead of + once per reader (fixes the O(readers × writers) re-parse). Both adapters + delegate to it. +- **Shared `validate_domain_id` (`common/dds_helpers`).** The identical 0..232 + bound check in all four DDS adapter constructors (Cyclone, Fast, OpenDDS, + Dust) is now defined once. +- **Cyclone discovery/sample reads switched from `take_iter` to `read_iter` + (audit A1/P1-5).** Destructive `take` drained the builtin discovery cache, + risking spurious lost / re-discovered participant flapping across polls; + `read` is non-destructive — the correct choice for read-only observability. + ⚠️ **Requires real-bus validation on `scripts/integration/` before release**: + the read-vs-take semantics cannot be exercised without `cyclonedds` installed + (the adapter is not importable in the unit environment; these edits are + validated only by `ruff` static analysis + `py_compile`). + +Baseline: 399 → 485 passed, 24 → 23 skipped, ruff clean, coverage 89.12%. + ## [0.5.0] - 2026-05-21 ### Sprint v0.5.0 — Polish + validation (pre-marketing-publication) diff --git a/README.md b/README.md index dcd4848..75c5da2 100644 --- a/README.md +++ b/README.md @@ -185,7 +185,7 @@ Six DDS / observability tools (in addition to the five ROS2 tools above) : **Composite adapter (v0.4.0 Phase 1+).** When `TOPICFORGE_MODE=live` is paired with a DDS backend (`cyclone`, `fast`, …), TopicForge instantiates **both** a ROS2 CLI adapter and the chosen DDS adapter and routes per-tool category — the 5 ROS2 tools hit the CLI, the DDS / observability tools hit the DDS backend. ROS2-only or DDS-only setups still work — the missing half is skipped and the present half serves what it can. The mock backend continues to expose all 11 tools against deterministic fixtures for local development. -**`peek_dds_samples` payload shape (v0.4.0 Phase 1.5).** Full-fidelity on the 4 builtin DCPS topics (`DCPSParticipant`, `DCPSSubscription`, `DCPSPublication`). Arbitrary user topics return best-effort decoded samples with a `_decode_status` annotation : `"full"` (every IDL field decoded — currently a v0.4.0+ Cyclone XTypes path), `"partial"` (some fields decoded, others opaque), or `"raw"` (binding could not resolve the dynamic XTypes — bytes preserved as hex in `_raw_bytes_hex`). The diagnostic key `_decode_note` carries a short explanation when the status is non-`full`. The wire shape is identical across Cyclone and Fast backends. +**`peek_dds_samples` payload shape (v0.4.0 Phase 1.5).** Full-fidelity on the 4 builtin DCPS topics (`DCPSParticipant`, `DCPSSubscription`, `DCPSPublication`). Arbitrary user topics return best-effort decoded samples with a `_decode_status` annotation : `"full"` (every IDL field decoded — currently a v0.4.0+ Cyclone XTypes path), `"partial"` (some fields decoded, others opaque), or `"raw"` (binding could not resolve the dynamic XTypes). The diagnostic key `_decode_note` carries a short explanation when the status is non-`full`. The wire shape is identical across Cyclone and Fast backends. **Caveat (v0.5.x):** on the current user-topic *raw* path `_raw_bytes_hex` is **empty** — a `"raw"` status means "topic present on the bus but not decoded", not "here are the serialized bytes to re-decode". Capturing the on-wire CDR bytes into the fallback is roadmapped ; until then use the Cyclone full/partial XTypes path for actual user-topic payloads. The 4 builtin DCPS topics are unaffected (always structured). **`RTI Connext`** is v0.4.0+ Pro tier (BYO license — see `docs/pro.md`). diff --git a/docs/product-plan.md b/docs/product-plan.md index 9d2e216..7ebe58a 100644 --- a/docs/product-plan.md +++ b/docs/product-plan.md @@ -8,7 +8,7 @@ TopicForge is **the safety-first read-only MCP for ROS2 robotics**. Where general-purpose ROS-MCP servers let an LLM publish topics, call services, and command robots — useful for demos, untenable for production fleets, defense systems, or anything safety-certified — TopicForge is read-only by **architecture**, not by configuration. There is no write path to misconfigure, no permission system to audit, no liability conversation to have. The MCP client can see the robot stack; it cannot touch it. -Concretely, the server exposes five typed tools today — `health_check`, `list_topics`, `get_topic_info`, `sample_messages`, `analyze_bag` — backed by either a deterministic mock adapter (no ROS2 required) or a `ros2` CLI wrapper (full live introspection). Outputs are frozen Pydantic schemas, stable across runtime modes. Telemetry is opt-in, six fields, zero user payload. +Concretely, the server exposes **eleven typed read-only tools today** (v0.5.0): the five ROS2-graph tools (`health_check`, `list_topics`, `get_topic_info`, `sample_messages`, `analyze_bag`) plus the six DDS / observability tools shipped across v0.2.0–v0.4.0 (`list_participants`, `detect_qos_mismatches`, `peek_dds_samples`, `participant_events`, `topic_metrics`, `peek_bag_samples`). They are backed by a deterministic mock adapter (no ROS2/DDS required), a `ros2` CLI wrapper, or an OSS DDS participant (Eclipse CycloneDDS / eProsima Fast DDS). Outputs are frozen Pydantic schemas, stable across runtime modes. Telemetry is opt-in, six fields, zero user payload. The ROS-MCP category is no longer empty (see §11 Risk register for the competitive landscape as of 2026-05-13). What TopicForge defends, and the rest of the pack will inherit, is the read-only-by-architecture stance and the production-quality engineering envelope around it — frozen schemas, mock-first development, telemetry contract pinned by tests, Windows-first cross-platform, no shell injection, deterministic outputs. @@ -40,13 +40,13 @@ Three concentric circles, ranked by strategic priority rather than acquisition c The strategic bet is **pack breadth via two focused products plus a modular surface inside TopicForge**. Two MCPs, not three to five — solo maintenance cost was the binding constraint and the 2026-05-14 audit collapsed the earlier 3-to-5-MCP plan accordingly. -**MCP 01 — TopicForge umbrella.** Covers ROS2 introspection today (shipped v0.1.2) and DDS observability as the next module (roadmapped, see §8). The `RosAdapter` protocol generalizes into a `MiddlewareAdapter` protocol that supports CycloneDDS in the OSS core and RTI Connext under the existing `topicforge_pro` license-gated package. One install (`pip install topicforge`, optional extras for DDS), one CLI, one license key — the umbrella keeps the developer ergonomics tight while extending coverage to the DDS-native audience the ROS-MCP competitive set does not reach. Module spec at `docs/projet-file/mcp-02-spec.md`. +**MCP 01 — TopicForge umbrella.** Covers ROS2 introspection (shipped v0.1.2) and DDS observability, which **shipped as a module across v0.2.0–v0.4.0** (see §8) — no longer roadmapped. The `RosAdapter` protocol was generalized into a `MiddlewareAdapter` protocol that supports Eclipse CycloneDDS and eProsima Fast DDS in the OSS core and RTI Connext under the existing `topicforge_pro` license-gated package. One install (`pip install topicforge`, optional extras for DDS), one CLI, one license key — the umbrella keeps the developer ergonomics tight while extending coverage to the DDS-native audience the ROS-MCP competitive set does not reach. Module spec at `docs/projet-file/mcp-02-spec.md`. **MCP 02 — DatasetForge.** Vision Dataset Inspector. Read images + annotations (COCO at MVP; YOLO / HF Datasets on roadmap) and answer structured questions about class balance, split coherence, annotation quality. Targets the ML/CV audience overlapping with TopicForge but distinct enough in domain (training data vs runtime graph) to warrant a separate product, separate repo, separate PyPI name. Full spec at `docs/projet-file/mcp-03-spec.md` (the file is still named `mcp-03-spec.md` for historical continuity ; the slot is MCP 02 of the 2-MCP pack). **Motif of the pivot.** Earlier drafts of this plan sequenced a 3-to-5-MCP pack with a separate DDS observability MCP as MCP 02. The 2026-05-14 audit collapsed that into a 2-product strategy: TopicForge as an umbrella covering both middlewares, DatasetForge as the second standalone product. The binding constraints were (a) solo-maintenance cost of running two repos in parallel and (b) the fact that ROS2 and DDS are the same problem shape — a typed pub/sub graph that needs structured introspection — and the `RosAdapter` protocol already generalizes to a `MiddlewareAdapter` superset with zero rework. Two products instead of three reduces the surface area without losing coverage. -The umbrella commits TopicForge to a slightly broader scope (cap: 5 ROS2 tools today + at most 3 DDS-side tools when the module ships, ceiling enforced by §11). The pack inherits the layer separation, mock-first development, opt-in telemetry, and read-only-by-architecture commitments from TopicForge. Pack-shared infrastructure extraction (telemetry, license, settings resolver into a `pack-template/` repo) becomes a non-decision at 2 products: fork-and-tweak from TopicForge to DatasetForge is acceptable ; revisit only if a third product is ever planned. +The umbrella commits TopicForge to a broader scope than first drafted: the DDS module shipped **six** DDS / observability tools across v0.2.0–v0.4.0 (not the three originally scoped), taking the surface to **11 tools total**. The original 8-tool ceiling was formally revised — see the re-scope decision in §11. The pack inherits the layer separation, mock-first development, opt-in telemetry, and read-only-by-architecture commitments from TopicForge. Pack-shared infrastructure extraction (telemetry, license, settings resolver into a `pack-template/` repo) becomes a non-decision at 2 products: fork-and-tweak from TopicForge to DatasetForge is acceptable ; revisit only if a third product is ever planned. --- @@ -171,7 +171,7 @@ The risks worth tracking explicitly. Updated 2026-05-13 with the competitive lan - **Cross-platform regressions on Windows.** TopicForge's primary developer environment is Windows. The Makefile uses POSIX shell syntax; users on plain PowerShell need the documented escape hatches. Mitigation: tested directly in CI on `ubuntu-latest` only today; Windows coverage is documented in `docs/TESTING.md` and exercised manually before each release. - **Telemetry trust.** Even opt-in telemetry can damage trust if the payload contract drifts. Mitigation: `tests/test_telemetry.py::test_payload_contains_only_whitelisted_keys` pins the six allowed keys. Any change requires a CHANGELOG entry and a README Telemetry section update in the same PR. - **Time / focus dilution.** A solo maintainer trying to drive two products (TopicForge umbrella + DatasetForge), a Pro tier inside each, marketing, and the DDS module on top of TopicForge is the realistic risk. The 2026-05-14 pivot from a 3-to-5-MCP pack to a 2-product strategy reduced the surface but did not eliminate the risk. Mitigation: explicit phase gates (do not start Phase 2 until Phase 1 is shipped, do not act on the DDS module marketing until Phase 2 has shipped) — though §8 schedules `MiddlewareAdapter` protocol prep during Phase 1. -- **Scope creep within the TopicForge umbrella.** Combining ROS2 + DDS introspection in one product risks bloating the tool surface beyond what a focused MCP should expose. Mitigation: tool surface stays capped at the 5 ROS2 tools today ; the DDS module adds at most 3 new tools (`list_participants`, `detect_qos_mismatches`, `peek_dds_samples`) when it ships. Any 9th tool needs an explicit re-scope discussion documented in this register before code lands. +- **Scope creep within the TopicForge umbrella.** Combining ROS2 + DDS introspection in one product risks bloating the tool surface beyond what a focused MCP should expose. **Re-scope decision (2026-07-08, ratified retroactively).** The register's original ceiling — 5 ROS2 tools + at most 3 DDS tools, any 9th tool gated on a re-scope discussion documented *here* before code lands — was crossed during v0.4.0 **without that discussion being recorded in this register**, a governance gap surfaced by the 2026-07-08 external audit. The three tools that broke it are deliberate and were acknowledged in the CHANGELOG and `docs/projet-file/mcp-02-spec.md §2` at ship time: `participant_events` (9th, v0.4.0 Phase 1), `topic_metrics` (10th, Phase 2), `peek_bag_samples` (11th, Phase 3). They are accepted; the revised ceiling is **11 tools**. A 12th tool now needs an explicit re-scope discussion documented in this register before code lands. Mitigation going forward: the `verify-change` skill's doc-drift step and the `docs-curator` sweep keep this register, `README.md`, and `CLAUDE.md` in sync so a ceiling break cannot ship undocumented again. --- diff --git a/docs/projet-file/action-plan-audit-2026-07-08.md b/docs/projet-file/action-plan-audit-2026-07-08.md new file mode 100644 index 0000000..69f7af4 --- /dev/null +++ b/docs/projet-file/action-plan-audit-2026-07-08.md @@ -0,0 +1,282 @@ +# Plan d'action — audit externe 2026-07-08 + +> Audit « regard neuf » (architecte externe, sans connaissance préalable) réalisé le 2026-07-08 +> sur la branche `chore/post-v0.5.0-cruft-sweep`. Baseline vérifiée : **399 passed, 24 skipped**, +> ruff clean, `~2.2 s`. Ce document traduit les constats en backlog priorisé **et** en boucles +> (« loops ») outillées, selon les bonnes pratiques Claude Code (turn-based / goal / time / proactive). +> +> Il complète — il ne remplace pas — `audit-post-v0.4.0.md` et `audit-followup-triage-v0.2.0.md`. + +--- + +## 1. Le constat central (à lire en premier) + +La suite est **verte, rapide, déterministe et de bonne qualité** — mais son vert **surreprésente la +confiance dans la couche DDS live**, qui est le cœur de la proposition de valeur marketing (« multi-vendor +OMG DDS-RTPS »). Trois faits se combinent : + +1. **Les deux adaptateurs DDS réels (`dds_cyclone` ~743 LOC, `dds_fast` ~712 LOC, ~35 % de `src/`) + ne sont jamais exécutés** : ils importent leur binding au niveau module, et les bindings + (`cyclonedds`, `fastdds`) sont absents en CI comme en local. Même leurs helpers purs + (normalisation QoS, extraction GUID/vendor) sont donc **inatteignables** par les tests. +2. **Le décodage des samples DDS sur topics utilisateur est non-fonctionnel en live** : + `_try_dynamic_decode_fast` renvoie toujours `None`, et le repli `annotate_raw(b"")` produit un + `_raw_bytes_hex` **toujours vide**. Le repli documenté (« récupérez les octets bruts ») est + impossible. +3. Le tooling de couverture (`pytest-cov`) **n'est pas installé** : aucune métrique de couverture + ne cadre ce trou. + +Conséquence : des bugs fonctionnels (fréquence `topic_metrics` fausse, gaps de séquence explosifs, +faux-négatif QoS Deadline) ont pu **shipper au vert**. La priorité n° 1 n'est pas un fix ponctuel, +c'est de **rendre la couche DDS testable** puis de la verrouiller. + +Le reste du code (ROS2 CLI, services, modèles, télémétrie, wiring, CI/CD, sécurité) est de **très +bonne facture** et confirme les deux promesses affichées : *read-only by architecture* et *Windows-first*. + +--- + +## 2. Backlog priorisé + +### P0 — À traiter avant toute publication marketing du volet DDS + +| ID | Constat | Fichier(s) | Action | +|----|---------|-----------|--------| +| P0-1 | Helpers DDS purs inatteignables → normalisation QoS (feeder du diagnostic phare) non testée | `dds_cyclone/adapter.py`, `dds_fast/adapter.py` | Extraire `_cyclone_qos_to_profile` / `_fast_qos_to_profile` / `_extract_guid/_vendor_id/_hostname/_topic_name` vers `adapters/common/` (comme `cdr_decoder`), puis `tests/test_dds_qos_normalization.py` sur objets factices duck-typés. | +| P0-2 | Décodage user-topic non-fonctionnel ; `_raw_bytes_hex` toujours vide | `dds_fast/adapter.py:492`, `dds_cyclone/adapter.py:551`, `cdr_decoder.py`, `xtypes.py` | Soit câbler les octets CDR réels dans le repli, soit **rétrograder honnêtement** README + descriptions d'outils (« payload user-topic pas encore disponible en live »). Ne pas laisser la prose promettre un repli impossible. | +| P0-3 | `pytest-cov` absent → aucun garde-fou de couverture | `pyproject.toml` (`[dev]`) | Ajouter `pytest-cov`, publier un seuil (au moins sur `services/`, `adapters/common/`, `config/`), afficher `term-missing`. | + +### P1 — Bugs fonctionnels d'outils livrés + +| ID | Constat | Fichier | Action | +|----|---------|---------|--------| +| P1-1 | `topic_metrics` : fréquence fausse (off-by-one **et** lignes de snapshot au même timestamp) | `metrics_buffer.py:157-164` + `_peek_builtin` des 2 adaptateurs | Fréquence sur instants d'arrivée distincts, `(N-1)/(newest-oldest)` ; ou masquer `frequency_hz_observed` pour les sources snapshot. | +| P1-2 | Gaps de séquence explosifs (wrap 16-bit, restart, multi-writer) | `metrics_buffer.py:217-237` | Grouper par writer GUID (ajouter le champ à `MetricsSample`) + détecter reset/wrap. | +| P1-3 | Faux-négatif QoS Deadline : reader fini + writer `None` (=infini) non signalé incompatible | `qos_analyzer.py:74-79` | Modéliser Deadline absente comme infinie dans la comparaison RxO. | +| P1-4 | `LifecycleBuffer._participants` non borné (fuite mémoire + bloat de sortie) | `lifecycle.py:59,193` | Cap LRU du dict / purge des `"left"` > fenêtre de rétention ; cap sur `snapshot_participants`. Corriger le docstring « bounded ». | +| P1-5 | Cyclone `take_iter` destructif + reader neuf par appel → flapping discovered/lost | `dds_cyclone/adapter.py` (217,242,332,366...) | Utiliser `read_iter` non-destructif ; reader builtin persistant par instance. | +| P1-6 | Dérive doc : `product-plan.md` §1/§4/§11 dit encore « five tools today » et sa **propre barrière de gouvernance** (« any 9th tool needs an explicit re-scope discussion documented in this register before code lands ») a été franchie sans réconciliation | `docs/product-plan.md` | Cascade docs-curator : aligner §1/§4/§5/§11/§13 sur les 11 outils livrés. Idem CLAUDE.md §2/§12 (même si local). | + +### P2 — Robustesse, dette, hygiène + +| ID | Constat | Fichier | Action | +|----|---------|---------|--------| +| P2-1 | Duplication ~40 % entre Cyclone et Fast (~250-350 LOC) | `dds_cyclone`, `dds_fast` | Base partagée `_DdsObservabilityBase` (validation domaine, 5 raisers ROS2, squelette `detect_qos_mismatches`, `participant_events`, `topic_metrics`, boucles metrics). | +| P2-2 | Stubs OpenDDS/Dust 95 % identiques ; `OpenDdsAdapter.is_available()` peut mentir (True alors que tout raise) | `dds_opendds`, `dds_dust` | Base `_StubAdapter(...)` ; `is_available()` OpenDDS → `False` tant que non implémenté. | +| P2-3 | `vendor` Literal (`cyclone/fast/rti/mock/unknown`) désaligné de la matrice 8 vendors | `models/schemas.py` (`ParticipantInfo`, `ParticipantEvent`) | Étendre le Literal (ou mapper vers `unknown`) avant que opendds/dust/opensplice/coredx/intercom émettent des participants. | +| P2-4 | I/O bag réelle jamais exécutée (rosbags pur-Python, absent) | CI | Ajouter `rosbags` à un extra CI pour lever le skip du seul test I/O réel. | +| P2-5 | Divers mineurs | — | `iter_field_names` sur `__slots__` string (`cdr_decoder.py:94`) ; `frequency_hz_declared` jamais peuplé ; `_encode_raw_bytes` hex-puis-tronque ; cap profondeur récursion `decode_field_value` ; `_KNOWN_TOOLS` sans `peek_bag_samples` ; import privé `_DDS_BACKEND_MODULES` ; dead code `parse_echo_yaml` / `_ = (...)`. | +| P2-6 | Durcissement hosted (déjà DEFER en triage) | inspector, ros2_live | `Path.resolve()` + racine autorisée pour bag ; rejet `-`-préfixé ; env-scrub subprocess. **À laisser DEFER tant que mono-tenant local.** | + +--- + +## 3. Stratégie de boucles (« loops ») par chantier + +Rappel du cadre (article Claude Code) : on choisit le type de boucle selon *ce qu'on délègue*. + +### 3.1 Turn-based + skill de vérification — pour les fixes de code (P0-2, P1-1..P1-5) + +Le point faible révélé par l'audit est **l'étape de vérification** : `make check` est vert alors que +la couche DDS est cassée. On encode donc « ce que veut dire *fait* » dans un skill, pour que Claude +s'auto-vérifie au lieu de se fier au vert. + +→ **Artefact créé : `.claude/skills/topicforge/verify-change/SKILL.md`** (voir §5). Il impose +`make check` **plus** : couverture sur les modules touchés, smoke-test mock-mode réel, et un +garde-fou explicite « vert ≠ DDS live vérifié ». + +### 3.2 Goal-based (`/goal`) — pour le chantier testabilité DDS (P0-1) + +Critère de sortie déterministe, idéal pour `/goal` : + +``` +/goal Extraire les helpers QoS/normalisation de dds_cyclone + dds_fast vers adapters/common/, + ajouter tests/test_dds_qos_normalization.py. Stop quand : les nouveaux tests passent, + pytest-cov montre >90 % de couverture sur le module extrait, et `make check` reste vert. + Stop après 6 itérations sinon remonter le blocage. +``` + +Même patron pour P0-3 (« stop quand `pytest --cov` tourne en CI avec un seuil publié »). + +### 3.3 Time-based (`/loop`, `/schedule`) — pour la dérive doc et le CI-watch + +- **Dérive doc (P1-6)** — récurrent, entrée qui change (compte d'outils, schémas). Cadence lente : + ``` + /schedule chaque lundi : lance le sub-agent docs-curator sur README, product-plan.md, CLAUDE.md, + CHANGELOG. /goal : zéro divergence de compte d'outils / de version / de schéma entre les + quatre. Ouvre un diff-only, ne publie rien. + ``` +- **CI-watch** (au moment d'une PR de fix) — événementiel plutôt que temporel : + ``` + /loop 5m vérifie la PR courante, corrige la CI qui casse (matrice 6 cellules Win/Linux × 3.11-3.13), + réponds aux commentaires de review. Stop quand la PR est verte et sans commentaire ouvert. + ``` + +### 3.4 Proactive — après publication (funnel bug reports) + +Une fois publié sur PyPI/marketplace, le flux « bug report » est récurrent et bien défini : +`/schedule` (triage) + `/goal` (« ne t'arrête pas tant que chaque report trouvé n'est pas trié, +actionné, répondu ») + workflow (explorer plusieurs fixes en worktrees parallèles, juge adversarial) ++ auto mode. **À n'activer qu'après P0/P1** — inutile d'automatiser le triage tant que la couche +observée est fausse. + +--- + +## 4. Stratégie modèle / tokens (patrons Fable 5) + +Routing **par complexité estimée**, pas par rôle figé — la majorité des tokens passent au tarif +worker, et on n'escalade que ce qui le mérite. Le tier worker n'est pas verrouillé sur Sonnet : +on monte sur Opus quand la tâche est sensible, on redescend sur Sonnet quand elle est mécanique. + +| Tier | Modèle | Pour quoi | Items du backlog | +|------|--------|-----------|------------------| +| Advisor / orchestrateur (rare, ~1 appel) | **Fable 5** | Décisions de correction délicates, arbitrages de contrat, design d'abstraction | Sémantique RxO Deadline (P1-3), design de la base partagée `_DdsObservabilityBase` (P2-1), arbitrage « câbler les octets CDR vs rétrograder la doc » (P0-2) | +| Worker complexe | **Opus 4.8** | Refactors protocol-sensibles, math à corriger avec soin, logique stateful | Extraction base + helpers (P0-1 / P2-1), fix fréquence + gaps `topic_metrics` (P1-1 / P1-2), bornage `LifecycleBuffer` (P1-4), `take_iter → read_iter` (P1-5) | +| Worker mécanique | **Sonnet 5** | Volumineux, déterministe, faible jugement | Cascade docs (P1-6), tests paramétrés une fois les helpers extraits (P0-1), collapse des stubs (P2-2), fixes mineurs (P2-5) | + +Règle : **estimer la complexité avant de déléguer** et router en conséquence. En cas de doute, un +appel unique à l'advisor Fable 5 pour cadrer, puis exécution au tier worker approprié. Chaque +sous-agent garde son cache → les appels répétés ne repaient pas le contexte partagé. + +- **Scripts > raisonnement** pour le déterministe : `make check`, le smoke-test mock-mode, le check + de dérive de compte d'outils (comparer `MVP_TOOLS` aux tables README/doc) = un script, pas un + raisonnement à chaque tour. +- **Piloter avant un grand run** : tout workflow multi-agents (ex. audit récurrent) se teste sur une + tranche avant de fan-out. + +--- + +## 5. Artefacts créés par cet audit + +1. **`docs/projet-file/action-plan-audit-2026-07-08.md`** — ce document (hors sdist via + `[tool.hatch.build.targets.sdist] exclude`). +2. **`.claude/skills/topicforge/verify-change/SKILL.md`** — skill de vérification audit-informé + (local, gitignored `.claude/` par design CLAUDE.md §13). + +Aucune modification de `src/` ni de tests dans le cadre de l'audit : les fixes ci-dessus sont du +backlog à valider par le mainteneur, pas des changements appliqués unilatéralement. + +--- + +## 6. Découpage en lots (exécution par batch) + +Chaque lot = une unité de travail cohérente, avec critère de sortie déterministe (`/goal`) et tier +modèle. Une branche `fix/lotN-...` par lot, un `make check` + skill `verify-change` vert avant merge. + +### Lot 0 — Socle testabilité *(prérequis dur — bloque tout le reste)* ✅ FAIT (2026-07-08) + +> Extraction faite vers `common/qos_normalize.py` + `common/dds_introspection.py` +> (behavior-preserving, alias back dans les adaptateurs). `pytest-cov` + `rosbags` +> ajoutés à `[dev]`, config coverage `fail_under=85`. 2 nouveaux fichiers de tests +> (modules extraits ~91-92 % couverts). Test bag I/O réel dé-skippé + réparé +> (drift API rosbags). **Baseline 399 → 457 passed, 24 → 23 skipped, ruff + cov verts.** + +- **Contenu** : P0-3 (installer `pytest-cov` + seuil publié) · P0-1 (extraire les helpers purs + `_cyclone/_fast_qos_to_profile` + `_extract_guid/_vendor_id/_hostname/_topic_name` vers + `adapters/common/` + `tests/test_dds_qos_normalization.py`) · P2-4 (ajouter `rosbags` à l'extra CI). +- **Pourquoi groupé** : rien de DDS n'est vérifiable sans ça. Sortir la logique dans `common/` rend + les fixes des lots suivants **single-homed et testés** (au lieu de dupliqués + aveugles). +- **Dépendance** : aucune. **Critère de sortie** : `pytest-cov` tourne, module extrait > 90 %, + nouveaux tests verts, `make check` vert, nombre de skips inchangé. +- **Tier** : Opus 4.8 (extraction) + Sonnet 5 (tests, config CI). **Effort** : M. + +### Lot 1 — Vérité documentaire *(indépendant, faible risque — insérable n'importe quand)* ✅ FAIT (2026-07-08) + +> `product-plan.md` §1/§4 réalignés sur 11 outils ; §11 : décision de re-scope +> rétroactive ajoutée qui **ferme la barrière de gouvernance franchie** (plafond +> révisé à 11, un 12ᵉ outil exige une discussion documentée). Honnêteté C1 : +> README + description `peek_dds_samples` ne promettent plus le repli +> `_raw_bytes_hex` (vide sur le chemin user-topic raw). **476 passed, verts.** +> *Non touché* : `CLAUDE.md` (gitignored/local, §2 « MVP verrouillé » possiblement +> gelé volontairement) — laissé au mainteneur, signalé dans le rapport. + +- **Contenu** : M6/P1-6 (aligner `product-plan.md` §1/4/11/13 + `CLAUDE.md` §2/12 sur les 11 outils, + fermer la barrière de gouvernance franchie) · volet doc de C1/P0-2 (aucune description d'outil ni + README ne promet un comportement live absent). +- **Dépendance** : aucune (l'arbitrage C1 « câbler vs rétrograder » = 1 appel advisor Fable 5). +- **Critère de sortie** : compte d'outils + version identiques dans README / product-plan / CLAUDE / + CHANGELOG / `MVP_TOOLS` ; zéro promesse doc non tenue. **Tier** : Fable 5 (arbitrage) → Sonnet 5 + (cascade `docs-curator`). **Effort** : S. + +### Lot 2 — Bugs fonctionnels metrics / QoS *(après Lot 0)* ✅ FAIT (2026-07-08) + +> P1-1 fréquence : `(N-1)/(newest-oldest)` sur le span réel des samples (plus de +> division par `now-oldest`) ; snapshot au même timestamp → `None`. P1-2 gaps : +> comptés par writer (nouveau `MetricsSample.writer_guid`) + garde reset/wrap à +> 10 000. P1-3 Deadline : absent = infini → reader fini vs writer absent = incompatible. +> Tests rouge-puis-vert ajoutés. **457 → 464 passed, ruff + cov (88.83 %) verts.** + +- **Contenu** : P1-1 (fréquence off-by-one + timestamps snapshot) · P1-2 (gaps par writer GUID + + détection wrap/reset) · P1-3 (faux-négatif Deadline RxO offered-infinite). +- **Pourquoi groupé** : trois outils livrés qui renvoient des chiffres faux ; tous en logique pure + désormais testable. **Critère de sortie** : chaque bug a un test rouge-puis-vert, math validée sur + distribution connue. **Tier** : Opus 4.8 + Fable 5 (sémantique RxO Deadline). **Effort** : M. + +### Lot 3 — Fiabilité runtime *(mémoire + lifecycle)* ✅ PARTIEL (2026-07-08) + +> **Fait (pur, testé)** : P1-4 `LifecycleBuffer._participants` borné à +> `MAX_PARTICIPANTS=4096` (évince les tombstones `"left"` d'abord) ; docstring +> « bounded » désormais vrai. `MetricsBuffer._samples` borné à `MAX_TOPICS=4096` +> (P2-5). Tests d'éviction ajoutés. **464 → 470 passed, cov 88.90 %.** +> **Reporté au batch rig** : P1-5 (`take_iter → read_iter`, reader persistant, +> anti-flapping) — non vérifiable sans bus réel, ne pas modifier l'adaptateur à +> l'aveugle. À traiter avec le Lot 5 sur `scripts/integration/`. + +- **Contenu** : P1-4 (borner `LifecycleBuffer._participants` + purge, corriger le docstring) · borner + `MetricsBuffer._samples` (P2-5) · P1-5 (`take_iter → read_iter`, reader persistant, anti-flapping). +- **Critère de sortie** : cap testé, docstring « bounded » vrai, pas de flapping sur scénario simulé. + ⚠️ **P1-5 non vérifiable sans bus réel** → valider sur le rig `scripts/integration/`. +- **Tier** : Opus 4.8. **Effort** : M. + +### Lot 4 — Nettoyage mineur *(mécanique, indépendant)* ✅ FAIT (2026-07-08) + +> OpenDDS `is_available()` → `False` (S1) · `__slots__` string (C2) · cap +> récursion `decode_field_value` (M6) · `_encode_raw_bytes` slice-avant-hex (M5) +> · test e2e `AdapterError → ToolError`/isError · test de cohérence vendor +> Literal (P2-3, pin — pas d'élargissement du contrat wire) · `_KNOWN_TOOLS` +> + `peek_bag_samples` · dead code retiré (bag_service). **470 → 476 passed, verts.** +> *Non fait (reporté)* : collapse des stubs OpenDDS/Dust (P2-2) — refactor à +> faible valeur, laissé pour plus tard. + +- **Contenu** : P2-3 (Literal `vendor` aligné) · S1 (`OpenDdsAdapter.is_available()` → `False`) · + `__slots__` string (`cdr_decoder.py:94`) · `_KNOWN_TOOLS` + `peek_bag_samples` · test e2e + `AdapterError → isError` · P2-2 (collapse stubs) · `_encode_raw_bytes` slice-avant-hex · cap + profondeur récursion · dead code (`parse_echo_yaml`, `_ = (...)`). +- **Critère de sortie** : `make check` vert, chaque mineur adressé ou justifié. **Tier** : Sonnet 5. + **Effort** : S-M. + +### Lot 5 — Déduplication base DDS *(risqué — en dernier)* ✅ PRÉPARÉ (2026-07-08) — ⚠️ à valider sur le rig + +> **Fait & testé (binding-free)** : logique de pairing `detect_qos_mismatches` +> extraite vers `common/qos_endpoints.py` (+ fix perf O(R*W) → profils writer +> pré-calculés) ; `validate_domain_id` partagé. Tests : `test_qos_endpoints.py` +> + validation domaine des stubs. **476 → 485 passed, cov 89.12 %.** +> **À valider sur le rig `scripts/integration/`** (non exécutable ici, bindings +> absents) : le wiring Cyclone/Fast de `detect_qos_mismatches` (validé ruff + +> py_compile seulement) et la bascule `take_iter → read_iter` (P1-5). Ne pas +> release avant un run réel-bus vert. +> *Non fait* : base-classe complète `_DdsObservabilityBase` (les 5 raisers +> ROS2 / participant_events / topic_metrics restent dupliqués, faible valeur, +> risque élevé sur code non testé) — la dedup à plus forte valeur (pairing QoS) +> est capturée. + +- **Contenu** : P2-1 (`_DdsObservabilityBase` : validation domaine, 5 raisers ROS2, squelette + `detect_qos_mismatches`, `participant_events`, `topic_metrics`, boucles metrics). +- **Pourquoi en dernier** : c'est le code le moins testable (adaptateurs à import-binding), donc le + refactor le plus risqué ; le faire **après** que la logique soit extraite en `common/` (Lot 0) + réduit sa surface. ⚠️ valider sur le rig intégration. **Tier** : Opus 4.8 + Fable 5 (design). + **Effort** : M-L. + +### Hors lots — DEFER +- **P2-6** (durcissement hosted : `Path.resolve()` + racine, rejet `-`-préfixé, env-scrub subprocess) + reste **DEFER** tant que le déploiement est mono-tenant local — conforme au triage existant. + +### Séquence recommandée + +``` +Lot 0 ──► Lot 2 ──► Lot 3 ──► Lot 5 + │ + └─► Lot 1 (indépendant, quand tu veux) + └─► Lot 4 (indépendant, quand tu veux) +``` + +Lot 0 d'abord (débloque). Lots 1 et 4 sont indépendants et peuvent s'insérer à tout moment. Lots 2 et +3 veulent Lot 0 fait. Lot 5 en dernier. Un lot par branche, mergé vert avant d'attaquer le suivant. diff --git a/pyproject.toml b/pyproject.toml index c0dbaad..a6da18d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -29,7 +29,13 @@ dependencies = [ [project.optional-dependencies] dev = [ "pytest>=8.0", + "pytest-cov>=5.0", "ruff>=0.4", + # rosbags (Apache-2.0, pure-Python) is a *dev* dependency so the + # bag-analysis I/O tests actually run in CI instead of self-skipping on + # `requires_rosbags`. It is NOT a runtime dependency — end users opt in + # via `topicforge[bags]`. + "rosbags>=0.9", ] # v0.3.0: split DDS extras per vendor. `[dds]` (the union) pulls both # OSS Python participants ; granular `[dds-cyclone]` / `[dds-fast]` @@ -107,6 +113,33 @@ markers = [ "requires_rosbags: tests needing the `rosbags` Python library; auto-skip without it (v0.4.0 Phase 3+)", ] +[tool.coverage.run] +branch = true +source = ["topicforge"] +omit = [ + # The two real DDS adapters import their vendor binding (cyclonedds / + # fastdds) at module top level, so they are structurally unreachable + # without those SDKs installed. Their pure logic was extracted into + # adapters/common/{dds_introspection,qos_normalize,cdr_decoder}.py + # (Lot 0) which IS measured here; the thin binding shells are exercised + # only under the `-m integration` real-bus tier. Excluded so the + # coverage gate reflects testable code rather than being dragged down + # by lines no unit run can reach. + "*/adapters/dds_cyclone/adapter.py", + "*/adapters/dds_fast/adapter.py", +] + +[tool.coverage.report] +show_missing = true +skip_covered = false +# Threshold reflects the unit-testable surface (the two binding shells are +# omitted above). Baseline at the Lot 0 achieved floor (87% total); the +# extracted pure modules (qos_normalize, dds_introspection) are ~100%, while +# factory.py / bag_service.py binding-present branches stay uncovered without +# the SDKs installed. Ratchet upward as those get tests; do not lower without +# a note in the changelog. +fail_under = 85 + [tool.ruff] line-length = 100 target-version = "py311" diff --git a/src/topicforge/adapters/common/__init__.py b/src/topicforge/adapters/common/__init__.py index 0c898f9..0e429e7 100644 --- a/src/topicforge/adapters/common/__init__.py +++ b/src/topicforge/adapters/common/__init__.py @@ -13,6 +13,18 @@ VendorTag, canonicalize_vendor_id, format_guid, + validate_domain_id, +) +from topicforge.adapters.common.dds_introspection import ( + cyclone_extract_guid, + cyclone_extract_hostname, + cyclone_extract_topic_name, + cyclone_extract_vendor_id, + fast_extract_guid, + fast_extract_hostname, + fast_extract_topic_name, + fast_extract_vendor_id, + is_removal, ) from topicforge.adapters.common.lifecycle import MAX_EVENTS, LifecycleBuffer from topicforge.adapters.common.metrics_buffer import ( @@ -21,6 +33,11 @@ MetricsSample, ) from topicforge.adapters.common.qos_analyzer import detect_mismatches +from topicforge.adapters.common.qos_endpoints import detect_mismatches_across_endpoints +from topicforge.adapters.common.qos_normalize import ( + cyclone_qos_to_profile, + fast_qos_to_profile, +) from topicforge.adapters.common.xtypes import ( DecodeStatus, annotate_full, @@ -41,12 +58,25 @@ "annotate_partial", "annotate_raw", "canonicalize_vendor_id", + "cyclone_extract_guid", + "cyclone_extract_hostname", + "cyclone_extract_topic_name", + "cyclone_extract_vendor_id", + "cyclone_qos_to_profile", "decode_dynamic_sample", "decode_field_value", "detect_mismatches", + "detect_mismatches_across_endpoints", "dynamic_type_name", "extract_publish_ns_from_payload", "extract_seq_from_payload", + "fast_extract_guid", + "fast_extract_hostname", + "fast_extract_topic_name", + "fast_extract_vendor_id", + "fast_qos_to_profile", "format_guid", + "is_removal", "iter_field_names", + "validate_domain_id", ] diff --git a/src/topicforge/adapters/common/cdr_decoder.py b/src/topicforge/adapters/common/cdr_decoder.py index 12ef37b..e162871 100644 --- a/src/topicforge/adapters/common/cdr_decoder.py +++ b/src/topicforge/adapters/common/cdr_decoder.py @@ -93,7 +93,10 @@ def iter_field_names(sample: Any) -> list[str]: return list(fields) slots = getattr(sample, "__slots__", None) if slots: - return list(slots) + # `__slots__` may legally be a bare string (a single slot name); + # list() on a string explodes it into characters, so wrap it first + # to avoid decoding one field as N garbage fields. (Audit C2.) + return [slots] if isinstance(slots, str) else list(slots) return ( [name for name in vars(sample) if not name.startswith("_")] if hasattr(sample, "__dict__") @@ -101,7 +104,14 @@ def iter_field_names(sample: Any) -> list[str]: ) -def decode_field_value(value: Any) -> object: +_MAX_DECODE_DEPTH = 32 +"""Recursion cap for `decode_field_value`. Beyond this depth a value is +collapsed to `repr()` rather than recursed into — guards against a +pathologically deep or self-referential decoded object graph raising +`RecursionError`. Normal DDS/ROS IDL types nest far shallower. (Audit M6.)""" + + +def decode_field_value(value: Any, *, _depth: int = 0) -> object: """Recursive decode of a single dynamic-type field value. Primitives and strings pass through. Sequences (list / tuple) are @@ -110,19 +120,26 @@ def decode_field_value(value: Any) -> object: recurse via the same field-iteration logic. Unsupported types (bytes, custom classes that resist iteration) collapse to their `repr()` so the payload remains JSON-serializable. + + `_depth` is internal — recursion beyond `_MAX_DECODE_DEPTH` collapses + to `repr()` to bound stack usage. """ + if _depth >= _MAX_DECODE_DEPTH: + return repr(value) if isinstance(value, (str, int, float, bool)) or value is None: return value if isinstance(value, (list, tuple)): - return [decode_field_value(v) for v in value] + return [decode_field_value(v, _depth=_depth + 1) for v in value] if isinstance(value, dict): - return {str(k): decode_field_value(v) for k, v in value.items()} + return {str(k): decode_field_value(v, _depth=_depth + 1) for k, v in value.items()} # Nested struct — recurse. if any(hasattr(value, attr) for attr in ("__dataclass_fields__", "__fields__", "__slots__")): nested: dict[str, object] = {} for field_name in iter_field_names(value): try: - nested[field_name] = decode_field_value(getattr(value, field_name)) + nested[field_name] = decode_field_value( + getattr(value, field_name), _depth=_depth + 1 + ) except Exception: # pragma: no cover nested[field_name] = f"" return nested diff --git a/src/topicforge/adapters/common/dds_helpers.py b/src/topicforge/adapters/common/dds_helpers.py index 8b96faa..a2ead54 100644 --- a/src/topicforge/adapters/common/dds_helpers.py +++ b/src/topicforge/adapters/common/dds_helpers.py @@ -16,6 +16,24 @@ from typing import Literal +from topicforge.adapters.base import AdapterError + +_DDS_DOMAIN_MIN = 0 +_DDS_DOMAIN_MAX = 232 + + +def validate_domain_id(domain_id: int) -> None: + """Raise `AdapterError` when `domain_id` is outside the DDS range 0..232. + + Shared by every DDS adapter constructor (Cyclone, Fast, OpenDDS, Dust) so + the bound check — and its exact message — is defined once. (Lot 5.) + """ + if domain_id < _DDS_DOMAIN_MIN or domain_id > _DDS_DOMAIN_MAX: + raise AdapterError( + f"domain_id must be in {_DDS_DOMAIN_MIN}..{_DDS_DOMAIN_MAX}, got {domain_id}" + ) + + VendorTag = Literal["cyclone", "fast", "rti", "mock", "unknown"] """Canonical vendor tag exposed on `ParticipantInfo.vendor`. diff --git a/src/topicforge/adapters/common/dds_introspection.py b/src/topicforge/adapters/common/dds_introspection.py new file mode 100644 index 0000000..26802ca --- /dev/null +++ b/src/topicforge/adapters/common/dds_introspection.py @@ -0,0 +1,184 @@ +"""Defensive field extraction from DDS discovery samples — binding-free. + +Extracted from the Cyclone and Fast adapters (Lot 0, audit 2026-07-08) so +the `getattr`-with-fallback sample introspection is unit-testable without +the `cyclonedds` / `fastdds` bindings installed. + +The two vendors expose subtly different discovery-sample shapes, so the +helpers stay **vendor-qualified** (`cyclone_*` / `fast_*`) and preserve each +adapter's exact behavior byte-for-byte — unifying them into a single set is +deliberately deferred to the Lot 5 adapter-dedup work, which the real-bus +integration rig can verify. Merging untested extraction paths blind (no +bindings here) is exactly the silent-regression risk the audit flagged. + +Every helper returns `None` / safe defaults rather than raising — a single +odd discovery sample must never break a whole tool call. +""" + +from __future__ import annotations + +from typing import Any + +# --------------------------------------------------------------------------- +# Cyclone variants +# --------------------------------------------------------------------------- + + +def cyclone_extract_guid(sample: Any) -> bytes | None: + """Pull the 16-byte GUID off a Cyclone discovery sample, if present.""" + for attr in ("key", "participant_key", "guid"): + v = getattr(sample, attr, None) + if v is None: + continue + if isinstance(v, bytes): + return v + inner = getattr(v, "value", None) + if isinstance(inner, bytes): + return inner + return None + + +def cyclone_extract_vendor_id(sample: Any) -> tuple[int, int] | None: + """Pull the 2-byte OMG vendor_id off a Cyclone discovery sample, if present.""" + v = getattr(sample, "vendor_id", None) + if v is None: + v = getattr(sample, "vendor", None) + if v is None: + return None + if isinstance(v, bytes) and len(v) >= 2: + return (v[0], v[1]) + inner = getattr(v, "vendorId", None) + if isinstance(inner, (bytes, tuple, list)) and len(inner) >= 2: + return (inner[0], inner[1]) + if isinstance(v, (tuple, list)) and len(v) >= 2: + return (v[0], v[1]) + return None + + +def cyclone_extract_hostname(sample: Any) -> str | None: + """Pull a hostname / participant-name hint off a Cyclone sample, if exposed.""" + for attr in ("hostname", "participant_name", "user_data"): + v = getattr(sample, attr, None) + if isinstance(v, (bytes, bytearray)): + try: + decoded = v.decode("utf-8", errors="replace") + except (UnicodeError, AttributeError): + continue + if decoded: + return decoded + if isinstance(v, str) and v: + return v + return None + + +def cyclone_extract_topic_name(sample: Any) -> str | None: + """Pull the topic name off a Cyclone endpoint sample (`topic_name` then `topic`).""" + v = getattr(sample, "topic_name", None) + if v is None: + v = getattr(sample, "topic", None) + if isinstance(v, str) and v: + return v + return None + + +# --------------------------------------------------------------------------- +# Fast DDS variants +# --------------------------------------------------------------------------- + + +def is_removal(status: Any) -> bool: + """Detect a 'participant/endpoint removed' discovery status across + binding versions. Fast DDS exposes status as either an enum value + or a string label — accept both. + """ + if status is None: + return False + s = str(status).upper() + return "REMOVED" in s or "DISPOSED" in s or "DROPPED" in s + + +def fast_extract_guid(sample: Any) -> bytes | None: + """Pull a 16-byte GUID off a Fast DDS discovery sample.""" + for attr in ("guid", "key", "participant_key"): + v = getattr(sample, attr, None) + if v is None: + continue + if isinstance(v, bytes): + return v + for inner_attr in ("value", "data", "guidPrefix"): + inner = getattr(v, inner_attr, None) + if isinstance(inner, bytes): + return inner + if isinstance(inner, (tuple, list)) and inner: + try: + return bytes(int(b) & 0xFF for b in inner) + except (TypeError, ValueError): + continue + if isinstance(v, (tuple, list)) and v: + try: + return bytes(int(b) & 0xFF for b in v) + except (TypeError, ValueError): + continue + return None + + +def fast_extract_vendor_id(sample: Any) -> tuple[int, int] | None: + """Pull the 2-byte OMG vendor_id off a Fast DDS discovery sample.""" + v = getattr(sample, "vendor_id", None) + if v is None: + info = getattr(sample, "info", None) + if info is not None: + v = getattr(info, "vendor_id", None) + if v is None: + return None + if isinstance(v, bytes) and len(v) >= 2: + return (v[0], v[1]) + if isinstance(v, (tuple, list)) and len(v) >= 2: + try: + return (int(v[0]), int(v[1])) + except (TypeError, ValueError): + return None + inner = getattr(v, "vendor_id", None) + if isinstance(inner, (bytes, tuple, list)) and len(inner) >= 2: + try: + return (int(inner[0]), int(inner[1])) + except (TypeError, ValueError): + return None + return None + + +def fast_extract_hostname(sample: Any) -> str | None: + """Pull a hostname / participant-name hint off a Fast DDS sample, if exposed.""" + for attr in ("hostname", "participant_name", "name", "user_data"): + v = getattr(sample, attr, None) + if isinstance(v, (bytes, bytearray)): + try: + decoded = v.decode("utf-8", errors="replace") + except (UnicodeError, AttributeError): + continue + if decoded: + return decoded + if isinstance(v, str) and v: + return v + return None + + +def fast_extract_topic_name(sample: Any) -> str | None: + """Pull the topic name off a Fast DDS endpoint sample (`topic_name` only).""" + v = getattr(sample, "topic_name", None) + if isinstance(v, str) and v: + return v + return None + + +__all__ = [ + "cyclone_extract_guid", + "cyclone_extract_hostname", + "cyclone_extract_topic_name", + "cyclone_extract_vendor_id", + "fast_extract_guid", + "fast_extract_hostname", + "fast_extract_topic_name", + "fast_extract_vendor_id", + "is_removal", +] diff --git a/src/topicforge/adapters/common/lifecycle.py b/src/topicforge/adapters/common/lifecycle.py index 5de92c9..bb4fb90 100644 --- a/src/topicforge/adapters/common/lifecycle.py +++ b/src/topicforge/adapters/common/lifecycle.py @@ -10,10 +10,13 @@ * **Pure logic at module level.** No DDS dependency. Tests pin behavior against synthetic input — same convention as `parse_topic_list` and `detect_mismatches` (the *"pure parsers / analyzers"* convention). -* **Bounded.** The event ring tops out at `MAX_EVENTS` (default 200) ; - overflow drops the oldest. Matches the +* **Bounded.** The event ring tops out at `MAX_EVENTS` (default 200) and + the participant map at `MAX_PARTICIPANTS` (default 4096) ; overflow drops + the oldest (tombstoned `"left"` participants first). Matches the `MAX_SAMPLE_COUNT=50` ergonomic of `sample_messages` — tools should - never return unbounded collections. + never return unbounded collections, and a long-running server on a churny + bus (each restarted node mints a fresh RTPS GUID) must not grow without + bound. * **Thread-safe.** Discovery callbacks fire on the underlying DDS library's worker thread (Fast) ; tool calls fire on the MCP request thread. An RLock guards every mutating method ; readers ( @@ -37,6 +40,11 @@ MAX_EVENTS = 200 """Hard cap on the event ring. Older entries drop out as new ones arrive.""" +MAX_PARTICIPANTS = 4096 +"""Hard cap on the number of distinct participants tracked. On overflow a +tombstoned (`status == "left"`) participant is dropped first, else the +oldest-inserted one — so a churny bus cannot grow the map without bound.""" + EventType = Literal["discovered", "lost"] EffectiveMode = Literal["mock", "live"] @@ -54,10 +62,13 @@ class LifecycleBuffer: filtering happens via the same clock. """ - def __init__(self, *, max_events: int = MAX_EVENTS) -> None: + def __init__( + self, *, max_events: int = MAX_EVENTS, max_participants: int = MAX_PARTICIPANTS + ) -> None: self._lock = threading.RLock() self._participants: dict[str, ParticipantInfo] = {} self._events: deque[ParticipantEvent] = deque(maxlen=max_events) + self._max_participants = max_participants # ------------------------- mutating operations -------------------------- @@ -82,6 +93,8 @@ def record_seen( with self._lock: existing = self._participants.get(guid) if existing is None: + if len(self._participants) >= self._max_participants: + self._evict_participant() self._participants[guid] = ParticipantInfo( guid=guid, vendor=vendor, @@ -226,6 +239,22 @@ def events_since( # --------------------------- private helpers ---------------------------- + def _evict_participant(self) -> None: + """Drop one participant to keep the map bounded. Called under lock. + + Prefers a tombstoned (`status == "left"`) entry so currently-active + participants survive ; falls back to the oldest-inserted entry + (dicts preserve insertion order) when every tracked participant is + still active. + """ + for guid, info in self._participants.items(): + if info.status == "left": + del self._participants[guid] + return + oldest = next(iter(self._participants), None) + if oldest is not None: + del self._participants[oldest] + def _append_event( self, *, diff --git a/src/topicforge/adapters/common/metrics_buffer.py b/src/topicforge/adapters/common/metrics_buffer.py index db5e635..53e9b6d 100644 --- a/src/topicforge/adapters/common/metrics_buffer.py +++ b/src/topicforge/adapters/common/metrics_buffer.py @@ -10,11 +10,13 @@ * **Pure logic at module level.** No DDS dependency. Tests pin behavior against synthetic input — same convention as `parse_topic_list`, `detect_mismatches`, `LifecycleBuffer`. -* **Bounded per-topic.** Each topic's ring caps at +* **Bounded per-topic and in topic count.** Each topic's ring caps at `MAX_SAMPLES_PER_TOPIC` (default 1000) ; older samples drop out - when new ones arrive. Memory footprint is bounded by - `O(topics x 1000 x sample_record_size)` — at 50 topics roughly - 10 MB worst case. + when new ones arrive. The number of distinct topics tracked caps at + `MAX_TOPICS` (default 4096), oldest-inserted evicted on overflow, so a + churny bus cannot grow the map without bound. Memory footprint is bounded + by `O(min(topics, 4096) x 1000 x sample_record_size)` — at 50 topics + roughly 10 MB worst case. * **Thread-safe.** Cyclone and Fast adapters today fill the buffer on the tool-call thread (synchronous), but a future rclpy adapter (roadmapped in `docs/product-plan.md §5`) will fire callbacks @@ -29,7 +31,7 @@ from __future__ import annotations import threading -from collections import deque +from collections import defaultdict, deque from dataclasses import dataclass from typing import Literal @@ -38,6 +40,10 @@ MAX_SAMPLES_PER_TOPIC = 1000 """Hard cap on per-topic ring buffer. Drop-oldest on overflow.""" +MAX_TOPICS = 4096 +"""Hard cap on the number of distinct topics tracked. Oldest-inserted topic +evicted on overflow so a churny bus cannot grow the map without bound.""" + EffectiveMode = Literal["mock", "live"] @@ -49,8 +55,10 @@ class MetricsSample: surfaces the sample. `receive_ns` is `time.time_ns()` at capture moment (wall clock, NOT DDS-RTPS receive timestamp — neither binding exposes the underlying RTPS timestamp through Python - reliably). `sequence_number` and `publish_ns` are best-effort — - `None` when the sample type doesn't expose them. + reliably). `sequence_number`, `publish_ns`, and `writer_guid` are + best-effort — `None` when the sample type / binding doesn't expose + them. `writer_guid` lets `compute_metrics` count sequence gaps per + writer instead of merging independent counters (Audit C6). """ topic: str @@ -58,14 +66,21 @@ class MetricsSample: sequence_number: int | None publish_ns: int | None domain_id: int + writer_guid: str | None = None class MetricsBuffer: """Per-topic bounded ring + percentile/frequency computation.""" - def __init__(self, *, max_samples_per_topic: int = MAX_SAMPLES_PER_TOPIC) -> None: + def __init__( + self, + *, + max_samples_per_topic: int = MAX_SAMPLES_PER_TOPIC, + max_topics: int = MAX_TOPICS, + ) -> None: self._lock = threading.RLock() self._cap = max_samples_per_topic + self._max_topics = max_topics self._samples: dict[str, deque[MetricsSample]] = {} # --------------------------- mutating ------------------------------ @@ -78,11 +93,17 @@ def record( sequence_number: int | None, publish_ns: int | None, domain_id: int, + writer_guid: str | None = None, ) -> None: """Append one sample to the per-topic ring. Oldest evicted on cap.""" with self._lock: ring = self._samples.get(topic) if ring is None: + if len(self._samples) >= self._max_topics: + # Evict the oldest-inserted topic to keep the map bounded. + oldest = next(iter(self._samples), None) + if oldest is not None: + del self._samples[oldest] ring = deque(maxlen=self._cap) self._samples[topic] = ring ring.append( @@ -92,6 +113,7 @@ def record( sequence_number=sequence_number, publish_ns=publish_ns, domain_id=domain_id, + writer_guid=writer_guid, ) ) @@ -154,18 +176,34 @@ def compute_metrics( # window_seconds_actual reflects the actual elapsed range # within the window — useful when the buffer is younger than # `window_seconds` (e.g., server just started). - oldest_ns = min(s.receive_ns for s in samples) + receive_times = [s.receive_ns for s in samples] + oldest_ns = min(receive_times) + newest_ns = max(receive_times) elapsed_ns = max(now_ns - oldest_ns, 1) # >=1 ns to avoid /0 window_actual_s = elapsed_ns / 1_000_000_000 - # A single sample doesn't define a frequency. + # Frequency is measured from the span of the samples' own arrival + # instants (newest - oldest) over N-1 intervals — NOT from + # (now - oldest), which would fold in idle time since the last peek. + # Samples surfaced by one opportunistic peek share a single + # receive_ns (span 0), so a snapshot legitimately yields no + # frequency rather than a fabricated rate. (Audit C5.) + sample_span_ns = newest_ns - oldest_ns freq_observed: float | None = ( - samples_observed / window_actual_s if samples_observed >= 2 else None + (samples_observed - 1) / (sample_span_ns / 1_000_000_000) + if samples_observed >= 2 and sample_span_ns > 0 + else None ) - seq_numbers = [s.sequence_number for s in samples if s.sequence_number is not None] - seq_available = len(seq_numbers) > 0 - gaps_count = _count_sequence_gaps(seq_numbers) if seq_available else 0 + # Sequence gaps are counted per writer: merging sequence numbers + # from independent writers on one topic would read each writer's + # counter offset as a huge phantom gap. (Audit C6.) + seq_by_writer: dict[str | None, list[int]] = defaultdict(list) + for s in samples: + if s.sequence_number is not None: + seq_by_writer[s.writer_guid].append(s.sequence_number) + seq_available = len(seq_by_writer) > 0 + gaps_count = sum(_count_sequence_gaps(seqs) for seqs in seq_by_writer.values()) latencies = [ s.receive_ns - s.publish_ns @@ -214,13 +252,25 @@ def sample_count(self, topic: str) -> int: # --------------------------------------------------------------------------- +# A hole wider than this between two consecutive observed sequence numbers is +# treated as a publisher restart / counter wrap (a discontinuity), not as that +# many genuinely lost samples — so a 16-bit wrap (65535→0) or a restart is not +# reported as tens of thousands of gaps. (Audit C6.) +_MAX_PLAUSIBLE_GAP = 10_000 + + def _count_sequence_gaps(seq_numbers: list[int]) -> int: - """Count missing entries in the observed sequence number list. + """Count missing entries in ONE writer's observed sequence numbers. + + Sorts + dedupes the input, then sums the holes between consecutive + values. Out-of-order arrivals are tolerated (we sort first) and + duplicates are removed. A single hole wider than `_MAX_PLAUSIBLE_GAP` + is treated as a reset/wrap discontinuity and skipped rather than + counted as that many losses. - Sorts the input and counts the gaps between consecutive values. - Out-of-order arrivals are tolerated (we sort first). Duplicates - are deduplicated before counting — they should not contribute to - a gap claim. + Callers pass one writer's sequence numbers — cross-writer merging is + handled in `compute_metrics` by grouping on writer GUID first, so an + independent writer's counter offset is never read as a phantom gap. Example: [0, 1, 2, 5, 6] → 2 gaps (3 and 4 missing). """ @@ -232,7 +282,7 @@ def _count_sequence_gaps(seq_numbers: list[int]) -> int: gaps = 0 for prev, curr in pairwise(unique): diff = curr - prev - if diff > 1: + if 1 < diff <= _MAX_PLAUSIBLE_GAP: gaps += diff - 1 return gaps diff --git a/src/topicforge/adapters/common/qos_analyzer.py b/src/topicforge/adapters/common/qos_analyzer.py index 1c3640e..2161aaa 100644 --- a/src/topicforge/adapters/common/qos_analyzer.py +++ b/src/topicforge/adapters/common/qos_analyzer.py @@ -26,6 +26,12 @@ "PERSISTENT", ) +# An absent Deadline QoS is the *infinite* default period (the loosest +# possible offer/request). Modeling `None` as +infinity lets the single +# RxO comparison below cover every case, including the reader-finite / +# writer-absent incompatibility the pre-audit code missed. +_INFINITE_DEADLINE = float("inf") + def detect_mismatches( reader_qos: QosProfile, writer_qos: QosProfile @@ -67,15 +73,21 @@ def detect_mismatches( if reader_qos.history == "KEEP_ALL" and writer_qos.history == "KEEP_LAST": risky.append("History") - # Deadline — a reader deadline strictly tighter than the writer - # deadline is incompatible: the writer cannot honor the reader's - # promise. If either side has no deadline (None), no constraint - # applies on that side. - if ( - reader_qos.deadline_ns is not None - and writer_qos.deadline_ns is not None - and reader_qos.deadline_ns < writer_qos.deadline_ns - ): + # Deadline — RxO rule: the writer's *offered* period must be <= the + # reader's *requested* period, else the writer cannot honor the reader's + # promise. An absent deadline is the infinite (loosest) default, so: + # * reader finite, writer finite → incompatible when writer > reader + # * reader finite, writer absent (∞) → incompatible (∞ > finite) ← the + # false negative the pre-audit code missed + # * reader absent (∞), writer anything → compatible (∞ requested) + # * both absent → compatible + reader_deadline = ( + reader_qos.deadline_ns if reader_qos.deadline_ns is not None else _INFINITE_DEADLINE + ) + writer_deadline = ( + writer_qos.deadline_ns if writer_qos.deadline_ns is not None else _INFINITE_DEADLINE + ) + if writer_deadline > reader_deadline: incompatible.append("Deadline") if not incompatible and not risky: diff --git a/src/topicforge/adapters/common/qos_endpoints.py b/src/topicforge/adapters/common/qos_endpoints.py new file mode 100644 index 0000000..28ba298 --- /dev/null +++ b/src/topicforge/adapters/common/qos_endpoints.py @@ -0,0 +1,95 @@ +"""Pair discovered reader/writer endpoints by topic and report QoS mismatches. + +Binding-free — extracted from the identical `detect_qos_mismatches` bodies of +`dds_cyclone/adapter.py` and `dds_fast/adapter.py` (Lot 5, audit 2026-07-08). +Both adapters had ~40 lines of the same "group endpoints by topic, pair each +reader against each writer, run the pure analyzer, build a `MismatchReport`" +logic — differing only in the vendor's `qos_to_profile` / `extract_*` +callables and the endpoint source. That logic now lives here, once, and is +unit-testable with synthetic endpoint objects (no `cyclonedds` / `fastdds`). + +Also fixes the O(readers x writers) QoS re-parse the audit flagged (P2/M7): +each writer's profile is computed once per topic, not once per reader. +""" + +from __future__ import annotations + +from collections.abc import Callable +from typing import Any + +from topicforge.adapters.common.dds_helpers import format_guid +from topicforge.adapters.common.qos_analyzer import detect_mismatches +from topicforge.models import MismatchReport, QosProfile + + +def detect_mismatches_across_endpoints( + *, + subs: list[Any], + pubs: list[Any], + topic: str | None, + qos_to_profile: Callable[[Any], QosProfile | None], + extract_topic_name: Callable[[Any], str | None], + extract_guid: Callable[[Any], bytes | None], + mode_effective: str = "live", +) -> list[MismatchReport]: + """Return one `MismatchReport` per incompatible (reader, writer) pair. + + `subs` / `pubs` are the discovered subscription / publication endpoint + samples (vendor-native shapes). `qos_to_profile`, `extract_topic_name`, + and `extract_guid` are the vendor's binding-free helpers (from + `common.qos_normalize` / `common.dds_introspection`). Pass `topic` to + scope to a single topic, or `None` for an exhaustive scan. + + Endpoints whose topic name cannot be resolved, or whose QoS cannot be + normalized to a `QosProfile`, are skipped — the analyzer needs a full + profile on both sides to make a meaningful claim. + """ + by_topic: dict[str, tuple[list[Any], list[Any]]] = {} + for sample in subs: + tname = extract_topic_name(sample) + if tname is None: + continue + if topic is not None and tname != topic: + continue + by_topic.setdefault(tname, ([], []))[0].append(sample) + for sample in pubs: + tname = extract_topic_name(sample) + if tname is None: + continue + if topic is not None and tname != topic: + continue + by_topic.setdefault(tname, ([], []))[1].append(sample) + + reports: list[MismatchReport] = [] + for tname, (readers, writers) in by_topic.items(): + # Precompute each writer's profile once per topic — the pre-audit code + # re-parsed every writer inside the reader loop (O(R*W)). (Audit P2/M7.) + writer_profiles: list[tuple[Any, QosProfile]] = [] + for writer_sample in writers: + writer_profile = qos_to_profile(writer_sample) + if writer_profile is not None: + writer_profiles.append((writer_sample, writer_profile)) + + for reader_sample in readers: + reader_profile = qos_to_profile(reader_sample) + if reader_profile is None: + continue + for writer_sample, writer_profile in writer_profiles: + result = detect_mismatches(reader_profile, writer_profile) + if result is None: + continue + policies, severity = result + reports.append( + MismatchReport( + topic=tname, + reader_guid=format_guid(extract_guid(reader_sample)), + writer_guid=format_guid(extract_guid(writer_sample)), + incompatible_policies=policies, + severity=severity, + mode_effective=mode_effective, # type: ignore[arg-type] + ) + ) + return reports + + +__all__ = ["detect_mismatches_across_endpoints"] diff --git a/src/topicforge/adapters/common/qos_normalize.py b/src/topicforge/adapters/common/qos_normalize.py new file mode 100644 index 0000000..bf2bbe8 --- /dev/null +++ b/src/topicforge/adapters/common/qos_normalize.py @@ -0,0 +1,178 @@ +"""Vendor QoS → canonical `QosProfile` normalization — binding-free, testable. + +Extracted from `dds_cyclone/adapter.py` and `dds_fast/adapter.py` +(Lot 0, audit 2026-07-08) so the QoS normalization that feeds +`detect_qos_mismatches` — the flagship DDS diagnostic — is unit-testable +**without** the `cyclonedds` / `fastdds` bindings installed. Previously +these functions lived below a top-level `import fastdds` / `from cyclonedds +...` in their adapters, so the entire QoS normalization path (and the bug +class where a renamed policy key silently returns `None` → no mismatch ever +reported) was unreachable by the test suite. + +Both adapters import these and alias them back to their original +`_cyclone_qos_to_profile` / `_fast_qos_to_profile` names, so their call +sites are unchanged. + +The Cyclone path keys policies by their binding class name (pure string +constants below). The Fast path keys by integer enum value, and those +integers come from the `fastdds` module — so `fast_qos_to_profile` takes +the three int→str maps as parameters (the adapter builds them from +`fastdds` and passes them in), keeping this module free of any binding +import. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from typing import Any + +from topicforge.models import QosProfile + +# CycloneDDS exposes QoS policies as instances of nested classes under +# `cyclonedds.qos.Policy.*` — we read them by simple class name to stay +# binding-version-agnostic. +CYCLONE_RELIABILITY_NAMES: dict[str, str] = {"Reliable": "RELIABLE", "BestEffort": "BEST_EFFORT"} +CYCLONE_DURABILITY_NAMES: dict[str, str] = { + "Volatile": "VOLATILE", + "TransientLocal": "TRANSIENT_LOCAL", + "Transient": "TRANSIENT", + "Persistent": "PERSISTENT", +} +CYCLONE_HISTORY_NAMES: dict[str, str] = {"KeepLast": "KEEP_LAST", "KeepAll": "KEEP_ALL"} + + +def cyclone_qos_to_profile(sample: Any) -> QosProfile | None: + """Map a Cyclone discovery sample's QoS into the canonical QosProfile. + + Returns `None` when essential QoS policies (reliability, durability, + history) are missing — the analyzer needs all three present to + produce a meaningful pair report. + """ + qos = getattr(sample, "qos", None) + if qos is None: + return None + + reliability: str | None = None + durability: str | None = None + history: str | None = None + history_depth: int | None = None + deadline_ns: int | None = None + + try: + for policy in qos: + cls_name = type(policy).__name__ + if cls_name in CYCLONE_RELIABILITY_NAMES: + reliability = CYCLONE_RELIABILITY_NAMES[cls_name] + elif cls_name in CYCLONE_DURABILITY_NAMES: + durability = CYCLONE_DURABILITY_NAMES[cls_name] + elif cls_name in CYCLONE_HISTORY_NAMES: + history = CYCLONE_HISTORY_NAMES[cls_name] + depth = getattr(policy, "depth", None) + if isinstance(depth, int): + history_depth = depth + elif cls_name == "Deadline": + d = getattr(policy, "duration", None) + if d is None: + d = getattr(policy, "deadline", None) + if hasattr(d, "to_nanoseconds"): + deadline_ns = int(d.to_nanoseconds()) + elif isinstance(d, int): + deadline_ns = d + except (TypeError, AttributeError): # defensive against odd qos shapes + return None + + if reliability is None or durability is None or history is None: + return None + + return QosProfile( + reliability=reliability, # type: ignore[arg-type] + durability=durability, # type: ignore[arg-type] + history=history, # type: ignore[arg-type] + history_depth=history_depth, + deadline_ns=deadline_ns, + ) + + +def fast_qos_to_profile( + sample: Any, + *, + reliability_map: Mapping[int, str], + durability_map: Mapping[int, str], + history_map: Mapping[int, str], +) -> QosProfile | None: + """Map a Fast DDS discovery sample's QoS into the canonical QosProfile. + + `reliability_map` / `durability_map` / `history_map` are the binding's + integer-enum → canonical-string tables. The adapter builds them from + `fastdds` constants and passes them in, so this function stays free of + any binding import and is testable with synthetic maps. + + Returns `None` when reliability, durability, or history cannot be + resolved — the analyzer needs all three. + """ + qos = getattr(sample, "qos", None) + if qos is None: + return None + + reliability: str | None = None + durability: str | None = None + history: str | None = None + history_depth: int | None = None + deadline_ns: int | None = None + + try: + rel = getattr(qos, "reliability", None) or getattr(qos, "m_reliability", None) + if rel is not None: + kind = getattr(rel, "kind", None) + if kind is not None: + reliability = reliability_map.get(kind) + + dur = getattr(qos, "durability", None) or getattr(qos, "m_durability", None) + if dur is not None: + kind = getattr(dur, "kind", None) + if kind is not None: + durability = durability_map.get(kind) + + hist = getattr(qos, "history", None) or getattr(qos, "m_history", None) + if hist is not None: + kind = getattr(hist, "kind", None) + if kind is not None: + history = history_map.get(kind) + depth = getattr(hist, "depth", None) + if isinstance(depth, int): + history_depth = depth + + ddl = getattr(qos, "deadline", None) or getattr(qos, "m_deadline", None) + if ddl is not None: + period = getattr(ddl, "period", None) + if period is not None: + sec = getattr(period, "seconds", None) + if sec is None: + sec = getattr(period, "sec", None) or 0 + nsec = getattr(period, "nanosec", None) + if nsec is None: + nsec = getattr(period, "nanoseconds", None) or 0 + if sec or nsec: + deadline_ns = int(sec) * 1_000_000_000 + int(nsec) + except (TypeError, AttributeError): # defensive + return None + + if reliability is None or durability is None or history is None: + return None + + return QosProfile( + reliability=reliability, # type: ignore[arg-type] + durability=durability, # type: ignore[arg-type] + history=history, # type: ignore[arg-type] + history_depth=history_depth, + deadline_ns=deadline_ns, + ) + + +__all__ = [ + "CYCLONE_DURABILITY_NAMES", + "CYCLONE_HISTORY_NAMES", + "CYCLONE_RELIABILITY_NAMES", + "cyclone_qos_to_profile", + "fast_qos_to_profile", +] diff --git a/src/topicforge/adapters/common/xtypes.py b/src/topicforge/adapters/common/xtypes.py index 343b75a..bced4d4 100644 --- a/src/topicforge/adapters/common/xtypes.py +++ b/src/topicforge/adapters/common/xtypes.py @@ -98,11 +98,14 @@ def annotate_raw(raw_bytes: bytes, *, note: str) -> dict[str, object]: def _encode_raw_bytes(raw_bytes: bytes) -> dict[str, object]: - """Encode bytes as hex with bounded length + truncation flag.""" - hex_str = raw_bytes.hex() - truncated = len(hex_str) > _RAW_BYTES_PREVIEW_LIMIT - if truncated: - hex_str = hex_str[:_RAW_BYTES_PREVIEW_LIMIT] + """Encode bytes as hex with bounded length + truncation flag. + + Slices the *bytes* before hex-encoding (each byte → 2 hex chars) so a + large payload does not allocate its full 2x-size hex string only to be + truncated to the preview limit. (Audit M5.) + """ + truncated = len(raw_bytes) * 2 > _RAW_BYTES_PREVIEW_LIMIT + hex_str = raw_bytes[: _RAW_BYTES_PREVIEW_LIMIT // 2].hex() if truncated else raw_bytes.hex() out: dict[str, object] = {"_raw_bytes_hex": hex_str} if truncated: out["_raw_bytes_truncated"] = True diff --git a/src/topicforge/adapters/dds_cyclone/adapter.py b/src/topicforge/adapters/dds_cyclone/adapter.py index 8511dee..af84313 100644 --- a/src/topicforge/adapters/dds_cyclone/adapter.py +++ b/src/topicforge/adapters/dds_cyclone/adapter.py @@ -65,12 +65,28 @@ canonicalize_vendor_id, decode_dynamic_sample, decode_field_value, - detect_mismatches, + detect_mismatches_across_endpoints, dynamic_type_name, extract_publish_ns_from_payload, extract_seq_from_payload, format_guid, iter_field_names, + validate_domain_id, +) +from topicforge.adapters.common import ( + cyclone_extract_guid as _extract_guid, +) +from topicforge.adapters.common import ( + cyclone_extract_hostname as _extract_hostname, +) +from topicforge.adapters.common import ( + cyclone_extract_topic_name as _extract_topic_name, +) +from topicforge.adapters.common import ( + cyclone_extract_vendor_id as _extract_vendor_id, +) +from topicforge.adapters.common import ( + cyclone_qos_to_profile as _cyclone_qos_to_profile, ) from topicforge.models import ( BagAnalysis, @@ -78,7 +94,6 @@ MismatchReport, ParticipantEvent, ParticipantInfo, - QosProfile, SampleResult, TopicInfo, TopicMetrics, @@ -100,8 +115,11 @@ log = logging.getLogger(__name__) # Tunables — kept module-level so a future env-var hook is a one-line -# change. Discovery is a bounded operation: `take_iter` returns whatever -# samples accumulated during the timeout window. +# change. Discovery + sample reads use `read_iter` (non-destructive) rather +# than `take_iter`, so observing the builtin discovery topics does not drain +# the reader cache and cause spurious lost / re-discovered participant +# flapping across polls. (Audit P1-5 — the read-vs-take semantics on a real +# bus must be confirmed on the `scripts/integration/` rig before this ships.) _DISCOVERY_TIMEOUT_SEC = 2.0 _SAMPLE_TIMEOUT_SEC = 1.0 _MAX_PARTICIPANTS = 256 @@ -214,7 +232,7 @@ def _discover_type_id_for_topic(dp: Any, topic: str) -> Any | None: """ try: reader = BuiltinDataReader(dp, BuiltinTopicDcpsPublication) - for sample in reader.take_iter(timeout=duration(seconds=_DISCOVERY_TIMEOUT_SEC)): + for sample in reader.read_iter(timeout=duration(seconds=_DISCOVERY_TIMEOUT_SEC)): if _extract_topic_name(sample) != topic: continue for attr in ("type_id", "type_identifier", "type_info"): @@ -239,7 +257,7 @@ def _collect_dynamic_samples(dp: Any, topic: str, type_object: Any, count: int) dynamic_topic = DynamicTopic(dp, topic, type_object) reader = DynamicDataReader(dp, dynamic_topic) - return list(reader.take_iter(timeout=duration(seconds=_SAMPLE_TIMEOUT_SEC)))[:count] + return list(reader.read_iter(timeout=duration(seconds=_SAMPLE_TIMEOUT_SEC)))[:count] except Exception: # pragma: no cover — binding-side error log.debug("typed reader construction failed for topic %r", topic, exc_info=True) return None @@ -268,8 +286,7 @@ class CycloneDdsAdapter: name: AdapterName = "cyclone" def __init__(self, domain_id: int = 0) -> None: - if domain_id < 0 or domain_id > 232: - raise AdapterError(f"domain_id must be in 0..232, got {domain_id}") + validate_domain_id(domain_id) self._domain_id = domain_id # v0.4.0 Phase 1: lifecycle tracking. Cyclone uses polling-delta # reconciliation — see `list_participants` for the feed pattern. @@ -329,7 +346,7 @@ def list_participants(self, domain_id: int = 0) -> list[ParticipantInfo]: """ try: reader = BuiltinDataReader(self._dp, BuiltinTopicDcpsParticipant) - samples = list(reader.take_iter(timeout=duration(seconds=_DISCOVERY_TIMEOUT_SEC))) + samples = list(reader.read_iter(timeout=duration(seconds=_DISCOVERY_TIMEOUT_SEC))) except Exception as exc: raise AdapterError( f"CycloneDDS participant discovery failed on domain {self._domain_id} " @@ -359,14 +376,20 @@ def list_participants(self, domain_id: int = 0) -> list[ParticipantInfo]: return self._lifecycle.snapshot_participants(domain_id=self._domain_id) def detect_qos_mismatches(self, topic: str | None = None) -> list[MismatchReport]: - """Pair reader/writer endpoints by topic, run the pure analyzer on each.""" + """Pair reader/writer endpoints by topic, run the shared analyzer on each. + + The pairing / reporting logic lives in + `common.qos_endpoints.detect_mismatches_across_endpoints` (shared with + the Fast adapter, unit-tested without a binding). This method only + gathers the vendor-native endpoint samples and hands them over. + """ try: sub_reader = BuiltinDataReader(self._dp, BuiltinTopicDcpsSubscription) pub_reader = BuiltinDataReader(self._dp, BuiltinTopicDcpsPublication) - subs = list(sub_reader.take_iter(timeout=duration(seconds=_DISCOVERY_TIMEOUT_SEC)))[ + subs = list(sub_reader.read_iter(timeout=duration(seconds=_DISCOVERY_TIMEOUT_SEC)))[ :_MAX_ENDPOINTS ] - pubs = list(pub_reader.take_iter(timeout=duration(seconds=_DISCOVERY_TIMEOUT_SEC)))[ + pubs = list(pub_reader.read_iter(timeout=duration(seconds=_DISCOVERY_TIMEOUT_SEC)))[ :_MAX_ENDPOINTS ] except Exception as exc: @@ -375,47 +398,14 @@ def detect_qos_mismatches(self, topic: str | None = None) -> list[MismatchReport f"({type(exc).__name__}: {exc})." ) from exc - by_topic: dict[str, tuple[list[Any], list[Any]]] = {} - for sample in subs: - tname = _extract_topic_name(sample) - if tname is None: - continue - if topic is not None and tname != topic: - continue - by_topic.setdefault(tname, ([], []))[0].append(sample) - for sample in pubs: - tname = _extract_topic_name(sample) - if tname is None: - continue - if topic is not None and tname != topic: - continue - by_topic.setdefault(tname, ([], []))[1].append(sample) - - reports: list[MismatchReport] = [] - for tname, (readers, writers) in by_topic.items(): - for reader_sample in readers: - reader_profile = _cyclone_qos_to_profile(reader_sample) - if reader_profile is None: - continue - for writer_sample in writers: - writer_profile = _cyclone_qos_to_profile(writer_sample) - if writer_profile is None: - continue - result = detect_mismatches(reader_profile, writer_profile) - if result is None: - continue - policies, severity = result - reports.append( - MismatchReport( - topic=tname, - reader_guid=format_guid(_extract_guid(reader_sample)), - writer_guid=format_guid(_extract_guid(writer_sample)), - incompatible_policies=policies, - severity=severity, - mode_effective="live", - ) - ) - return reports + return detect_mismatches_across_endpoints( + subs=subs, + pubs=pubs, + topic=topic, + qos_to_profile=_cyclone_qos_to_profile, + extract_topic_name=_extract_topic_name, + extract_guid=_extract_guid, + ) def peek_dds_samples(self, topic: str, count: int) -> SampleResult: """Peek recent samples on a DDS topic. @@ -442,7 +432,7 @@ def _peek_builtin(self, topic: str, count: int) -> SampleResult: topic_class = _BUILTIN_DCPS_TOPICS[topic] try: reader = BuiltinDataReader(self._dp, topic_class) - samples_raw = list(reader.take_iter(timeout=duration(seconds=_SAMPLE_TIMEOUT_SEC)))[ + samples_raw = list(reader.read_iter(timeout=duration(seconds=_SAMPLE_TIMEOUT_SEC)))[ :count ] except Exception as exc: @@ -566,10 +556,10 @@ def _is_topic_on_bus(self, topic: str) -> bool: try: sub_reader = BuiltinDataReader(self._dp, BuiltinTopicDcpsSubscription) pub_reader = BuiltinDataReader(self._dp, BuiltinTopicDcpsPublication) - subs = list(sub_reader.take_iter(timeout=duration(seconds=_DISCOVERY_TIMEOUT_SEC)))[ + subs = list(sub_reader.read_iter(timeout=duration(seconds=_DISCOVERY_TIMEOUT_SEC)))[ :_MAX_ENDPOINTS ] - pubs = list(pub_reader.take_iter(timeout=duration(seconds=_DISCOVERY_TIMEOUT_SEC)))[ + pubs = list(pub_reader.read_iter(timeout=duration(seconds=_DISCOVERY_TIMEOUT_SEC)))[ :_MAX_ENDPOINTS ] except Exception: # pragma: no cover — defensive @@ -615,129 +605,11 @@ def topic_metrics( ) -# --------------------------------------------------------------------------- -# Sample-introspection helpers — defensive against binding shape variations. -# Each helper returns None / "unknown" / safe defaults rather than raising, -# so a single odd discovery sample never breaks the whole tool call. -# --------------------------------------------------------------------------- - - -def _extract_guid(sample: Any) -> bytes | None: - """Pull the 16-byte GUID off a discovery sample, if present.""" - for attr in ("key", "participant_key", "guid"): - v = getattr(sample, attr, None) - if v is None: - continue - if isinstance(v, bytes): - return v - inner = getattr(v, "value", None) - if isinstance(inner, bytes): - return inner - return None - - -def _extract_vendor_id(sample: Any) -> tuple[int, int] | None: - """Pull the 2-byte OMG vendor_id off a discovery sample, if present.""" - v = getattr(sample, "vendor_id", None) - if v is None: - v = getattr(sample, "vendor", None) - if v is None: - return None - if isinstance(v, bytes) and len(v) >= 2: - return (v[0], v[1]) - inner = getattr(v, "vendorId", None) - if isinstance(inner, (bytes, tuple, list)) and len(inner) >= 2: - return (inner[0], inner[1]) - if isinstance(v, (tuple, list)) and len(v) >= 2: - return (v[0], v[1]) - return None - - -def _extract_hostname(sample: Any) -> str | None: - """Pull a hostname / participant-name hint off a sample, if exposed.""" - for attr in ("hostname", "participant_name", "user_data"): - v = getattr(sample, attr, None) - if isinstance(v, (bytes, bytearray)): - try: - decoded = v.decode("utf-8", errors="replace") - except (UnicodeError, AttributeError): - continue - if decoded: - return decoded - if isinstance(v, str) and v: - return v - return None - - -def _extract_topic_name(sample: Any) -> str | None: - v = getattr(sample, "topic_name", None) - if v is None: - v = getattr(sample, "topic", None) - if isinstance(v, str) and v: - return v - return None - - -# QoS Policy class-name → canonical string maps. CycloneDDS exposes -# policies as instances of nested classes under `cyclonedds.qos.Policy.*` -# — we read them by simple class name to stay binding-version-agnostic. -_RELIABILITY_NAMES = {"Reliable": "RELIABLE", "BestEffort": "BEST_EFFORT"} -_DURABILITY_NAMES = { - "Volatile": "VOLATILE", - "TransientLocal": "TRANSIENT_LOCAL", - "Transient": "TRANSIENT", - "Persistent": "PERSISTENT", -} -_HISTORY_NAMES = {"KeepLast": "KEEP_LAST", "KeepAll": "KEEP_ALL"} - - -def _cyclone_qos_to_profile(sample: Any) -> QosProfile | None: - """Map a Cyclone discovery sample's QoS into the canonical QosProfile. - - Returns `None` when essential QoS policies (reliability, durability, - history) are missing — the analyzer needs all three present to - produce a meaningful pair report. - """ - qos = getattr(sample, "qos", None) - if qos is None: - return None - - reliability: str | None = None - durability: str | None = None - history: str | None = None - history_depth: int | None = None - deadline_ns: int | None = None - - try: - for policy in qos: - cls_name = type(policy).__name__ - if cls_name in _RELIABILITY_NAMES: - reliability = _RELIABILITY_NAMES[cls_name] - elif cls_name in _DURABILITY_NAMES: - durability = _DURABILITY_NAMES[cls_name] - elif cls_name in _HISTORY_NAMES: - history = _HISTORY_NAMES[cls_name] - depth = getattr(policy, "depth", None) - if isinstance(depth, int): - history_depth = depth - elif cls_name == "Deadline": - d = getattr(policy, "duration", None) - if d is None: - d = getattr(policy, "deadline", None) - if hasattr(d, "to_nanoseconds"): - deadline_ns = int(d.to_nanoseconds()) - elif isinstance(d, int): - deadline_ns = d - except (TypeError, AttributeError): # defensive against odd qos shapes - return None - - if reliability is None or durability is None or history is None: - return None - - return QosProfile( - reliability=reliability, # type: ignore[arg-type] - durability=durability, # type: ignore[arg-type] - history=history, # type: ignore[arg-type] - history_depth=history_depth, - deadline_ns=deadline_ns, - ) +# Sample-introspection helpers (_extract_guid / _extract_vendor_id / +# _extract_hostname / _extract_topic_name) and the QoS normalizer +# (_cyclone_qos_to_profile) were moved to the binding-free +# `topicforge.adapters.common.dds_introspection` / +# `.qos_normalize` modules (Lot 0, audit 2026-07-08) so they are +# unit-testable without the cyclonedds bindings installed. They are +# imported and aliased back to their original names at the top of this +# module, so the call sites above are unchanged. diff --git a/src/topicforge/adapters/dds_dust/adapter.py b/src/topicforge/adapters/dds_dust/adapter.py index 724f3f9..bc7b4f3 100644 --- a/src/topicforge/adapters/dds_dust/adapter.py +++ b/src/topicforge/adapters/dds_dust/adapter.py @@ -18,6 +18,7 @@ import logging from topicforge.adapters.base import AdapterError, AdapterName, EffectiveMode +from topicforge.adapters.common import validate_domain_id from topicforge.models import ( BagAnalysis, MessageSample, @@ -47,8 +48,7 @@ class DustDdsAdapter: name: AdapterName = "dust" def __init__(self, domain_id: int = 0) -> None: - if domain_id < 0 or domain_id > 232: - raise AdapterError(f"domain_id must be in 0..232, got {domain_id}") + validate_domain_id(domain_id) self._domain_id = domain_id @property diff --git a/src/topicforge/adapters/dds_fast/adapter.py b/src/topicforge/adapters/dds_fast/adapter.py index b6e9777..495f453 100644 --- a/src/topicforge/adapters/dds_fast/adapter.py +++ b/src/topicforge/adapters/dds_fast/adapter.py @@ -55,8 +55,27 @@ MetricsBuffer, annotate_raw, canonicalize_vendor_id, - detect_mismatches, + detect_mismatches_across_endpoints, format_guid, + validate_domain_id, +) +from topicforge.adapters.common import ( + fast_extract_guid as _extract_guid, +) +from topicforge.adapters.common import ( + fast_extract_hostname as _extract_hostname, +) +from topicforge.adapters.common import ( + fast_extract_topic_name as _extract_topic_name, +) +from topicforge.adapters.common import ( + fast_extract_vendor_id as _extract_vendor_id, +) +from topicforge.adapters.common import ( + fast_qos_to_profile as _common_fast_qos_to_profile, +) +from topicforge.adapters.common import ( + is_removal as _is_removal, ) from topicforge.models import ( BagAnalysis, @@ -191,8 +210,7 @@ def __init__( *, discovery_wait_ms: int = _DEFAULT_DISCOVERY_WAIT_MS, ) -> None: - if domain_id < 0 or domain_id > 232: - raise AdapterError(f"domain_id must be in 0..232, got {domain_id}") + validate_domain_id(domain_id) self._domain_id = domain_id # v0.4.0 Phase 1: lifecycle buffer fed by listener callbacks. self._lifecycle = LifecycleBuffer() @@ -284,50 +302,21 @@ def list_participants(self, domain_id: int = 0) -> list[ParticipantInfo]: return self._lifecycle.snapshot_participants(domain_id=self._domain_id) def detect_qos_mismatches(self, topic: str | None = None) -> list[MismatchReport]: - subs = self._listener.snapshot_subscriptions() - pubs = self._listener.snapshot_publications() - - by_topic: dict[str, tuple[list[Any], list[Any]]] = {} - for sample in subs: - tname = _extract_topic_name(sample) - if tname is None: - continue - if topic is not None and tname != topic: - continue - by_topic.setdefault(tname, ([], []))[0].append(sample) - for sample in pubs: - tname = _extract_topic_name(sample) - if tname is None: - continue - if topic is not None and tname != topic: - continue - by_topic.setdefault(tname, ([], []))[1].append(sample) - - reports: list[MismatchReport] = [] - for tname, (readers, writers) in by_topic.items(): - for reader_sample in readers: - reader_profile = _fast_qos_to_profile(reader_sample) - if reader_profile is None: - continue - for writer_sample in writers: - writer_profile = _fast_qos_to_profile(writer_sample) - if writer_profile is None: - continue - result = detect_mismatches(reader_profile, writer_profile) - if result is None: - continue - policies, severity = result - reports.append( - MismatchReport( - topic=tname, - reader_guid=format_guid(_extract_guid(reader_sample)), - writer_guid=format_guid(_extract_guid(writer_sample)), - incompatible_policies=policies, - severity=severity, - mode_effective="live", - ) - ) - return reports + """Pair reader/writer endpoints by topic via the shared analyzer. + + The pairing / reporting logic lives in + `common.qos_endpoints.detect_mismatches_across_endpoints` (shared with + the Cyclone adapter, unit-tested without a binding). This method only + supplies the listener's discovery snapshots and the Fast helpers. + """ + return detect_mismatches_across_endpoints( + subs=self._listener.snapshot_subscriptions(), + pubs=self._listener.snapshot_publications(), + topic=topic, + qos_to_profile=_fast_qos_to_profile, + extract_topic_name=_extract_topic_name, + extract_guid=_extract_guid, + ) def peek_dds_samples(self, topic: str, count: int) -> SampleResult: """v0.4.0 Phase 1: builtin DCPS snapshots + user-topic raw fallback. @@ -533,93 +522,17 @@ def _try_dynamic_decode_fast(topic: str, count: int) -> list[MessageSample] | No return None -# --------------------------------------------------------------------------- -# Internal helpers — defensive against binding shape variations across -# Fast DDS Python binding versions. Same convention as the Cyclone helpers: -# never raise, collapse missing data to None / "unknown" / safe defaults. -# --------------------------------------------------------------------------- - - -def _is_removal(status: Any) -> bool: - """Detect a 'participant/endpoint removed' discovery status across - binding versions. Fast DDS exposes status as either an enum value - or a string label — accept both. - """ - if status is None: - return False - s = str(status).upper() - return "REMOVED" in s or "DISPOSED" in s or "DROPPED" in s - - -def _extract_guid(sample: Any) -> bytes | None: - """Pull a 16-byte GUID off a Fast DDS discovery sample.""" - for attr in ("guid", "key", "participant_key"): - v = getattr(sample, attr, None) - if v is None: - continue - if isinstance(v, bytes): - return v - for inner_attr in ("value", "data", "guidPrefix"): - inner = getattr(v, inner_attr, None) - if isinstance(inner, bytes): - return inner - if isinstance(inner, (tuple, list)) and inner: - try: - return bytes(int(b) & 0xFF for b in inner) - except (TypeError, ValueError): - continue - if isinstance(v, (tuple, list)) and v: - try: - return bytes(int(b) & 0xFF for b in v) - except (TypeError, ValueError): - continue - return None - - -def _extract_vendor_id(sample: Any) -> tuple[int, int] | None: - v = getattr(sample, "vendor_id", None) - if v is None: - info = getattr(sample, "info", None) - if info is not None: - v = getattr(info, "vendor_id", None) - if v is None: - return None - if isinstance(v, bytes) and len(v) >= 2: - return (v[0], v[1]) - if isinstance(v, (tuple, list)) and len(v) >= 2: - try: - return (int(v[0]), int(v[1])) - except (TypeError, ValueError): - return None - inner = getattr(v, "vendor_id", None) - if isinstance(inner, (bytes, tuple, list)) and len(inner) >= 2: - try: - return (int(inner[0]), int(inner[1])) - except (TypeError, ValueError): - return None - return None - - -def _extract_hostname(sample: Any) -> str | None: - for attr in ("hostname", "participant_name", "name", "user_data"): - v = getattr(sample, attr, None) - if isinstance(v, (bytes, bytearray)): - try: - decoded = v.decode("utf-8", errors="replace") - except (UnicodeError, AttributeError): - continue - if decoded: - return decoded - if isinstance(v, str) and v: - return v - return None - - -def _extract_topic_name(sample: Any) -> str | None: - v = getattr(sample, "topic_name", None) - if isinstance(v, str) and v: - return v - return None +# Sample-introspection helpers (_is_removal / _extract_guid / +# _extract_vendor_id / _extract_hostname / _extract_topic_name) were moved +# to the binding-free `topicforge.adapters.common.dds_introspection` module +# (Lot 0, audit 2026-07-08) so they are unit-testable without the fastdds +# bindings installed. They are imported and aliased back to their original +# names at the top of this module, so the call sites above are unchanged. +# +# The QoS enum maps below stay here because they read integer values from +# the `fastdds` binding itself. The normalization logic that consumes them +# lives in `common.qos_normalize.fast_qos_to_profile` (also testable with +# synthetic maps) — `_fast_qos_to_profile` below binds the two together. # QoS enum integer values come from the binding's own constants rather @@ -653,60 +566,16 @@ def _build_history_map() -> dict[int, str]: def _fast_qos_to_profile(sample: Any) -> QosProfile | None: - qos = getattr(sample, "qos", None) - if qos is None: - return None - - reliability: str | None = None - durability: str | None = None - history: str | None = None - history_depth: int | None = None - deadline_ns: int | None = None + """Bind the shared Fast normalizer to this binding's enum maps. - try: - rel = getattr(qos, "reliability", None) or getattr(qos, "m_reliability", None) - if rel is not None: - kind = getattr(rel, "kind", None) - if kind is not None: - reliability = _RELIABILITY_MAP.get(kind) - - dur = getattr(qos, "durability", None) or getattr(qos, "m_durability", None) - if dur is not None: - kind = getattr(dur, "kind", None) - if kind is not None: - durability = _DURABILITY_MAP.get(kind) - - hist = getattr(qos, "history", None) or getattr(qos, "m_history", None) - if hist is not None: - kind = getattr(hist, "kind", None) - if kind is not None: - history = _HISTORY_MAP.get(kind) - depth = getattr(hist, "depth", None) - if isinstance(depth, int): - history_depth = depth - - ddl = getattr(qos, "deadline", None) or getattr(qos, "m_deadline", None) - if ddl is not None: - period = getattr(ddl, "period", None) - if period is not None: - sec = getattr(period, "seconds", None) - if sec is None: - sec = getattr(period, "sec", None) or 0 - nsec = getattr(period, "nanosec", None) - if nsec is None: - nsec = getattr(period, "nanoseconds", None) or 0 - if sec or nsec: - deadline_ns = int(sec) * 1_000_000_000 + int(nsec) - except (TypeError, AttributeError): # defensive - return None - - if reliability is None or durability is None or history is None: - return None - - return QosProfile( - reliability=reliability, # type: ignore[arg-type] - durability=durability, # type: ignore[arg-type] - history=history, # type: ignore[arg-type] - history_depth=history_depth, - deadline_ns=deadline_ns, + The pure normalization logic lives in + `common.qos_normalize.fast_qos_to_profile` ; this thin wrapper feeds it + the `fastdds`-derived int→str maps so the call sites in + `detect_qos_mismatches` stay unchanged. + """ + return _common_fast_qos_to_profile( + sample, + reliability_map=_RELIABILITY_MAP, + durability_map=_DURABILITY_MAP, + history_map=_HISTORY_MAP, ) diff --git a/src/topicforge/adapters/dds_opendds/adapter.py b/src/topicforge/adapters/dds_opendds/adapter.py index c959a6c..0ce0cf1 100644 --- a/src/topicforge/adapters/dds_opendds/adapter.py +++ b/src/topicforge/adapters/dds_opendds/adapter.py @@ -2,11 +2,16 @@ `pyopendds` is not currently maintained on PyPI. This stub implements the full `MiddlewareAdapter` protocol so the auto-detect framework -treats OpenDDS uniformly with Cyclone / Fast / RTI : the constructor -probes the binding via `importlib.util.find_spec("pyopendds")` ; -`is_available()` returns False when the binding cannot be found ; the -8 protocol methods raise `AdapterError(_OPENDDS_ROADMAP_MSG)` with a -clear pointer to the v0.5+ roadmap if a user reaches them. +treats OpenDDS uniformly with Cyclone / Fast / RTI ; the 8 protocol +methods raise `AdapterError(_OPENDDS_ROADMAP_MSG)` with a clear pointer +to the v0.5+ roadmap if a user reaches them. + +`is_available()` always returns False while this is a stub (Audit S1): +the factory selects backends on `is_available()`, so reporting True — +even when a `pyopendds` module happens to be importable — would make the +factory pick OpenDDS and then every tool call would raise. The Dust stub +follows the same rule. When a real binding ships, the replacement adapter +sets this from an actual capability probe. When `pyopendds` ships, the stub is replaced by a real adapter under the same module path. No other code changes : the factory, the @@ -18,10 +23,10 @@ from __future__ import annotations -import importlib.util import logging from topicforge.adapters.base import AdapterError, AdapterName, EffectiveMode +from topicforge.adapters.common import validate_domain_id from topicforge.models import ( BagAnalysis, MessageSample, @@ -52,17 +57,18 @@ class OpenDdsAdapter: name: AdapterName = "opendds" def __init__(self, domain_id: int = 0) -> None: - if domain_id < 0 or domain_id > 232: - raise AdapterError(f"domain_id must be in 0..232, got {domain_id}") + validate_domain_id(domain_id) self._domain_id = domain_id - self._binding_available = importlib.util.find_spec("pyopendds") is not None @property def effective_mode(self) -> EffectiveMode: return "live" def is_available(self) -> bool: - return self._binding_available + # Always False while this is a stub — even if a `pyopendds` module is + # importable, this adapter cannot serve any request, and the factory + # selects on is_available(). (Audit S1.) + return False # ----- ROS2 surface: not served by this adapter ----- diff --git a/src/topicforge/services/bag_service.py b/src/topicforge/services/bag_service.py index b4c38c0..80c2ea4 100644 --- a/src/topicforge/services/bag_service.py +++ b/src/topicforge/services/bag_service.py @@ -26,11 +26,7 @@ from typing import Any from topicforge.adapters.base import AdapterError -from topicforge.adapters.common import ( - annotate_full, - annotate_partial, - annotate_raw, -) +from topicforge.adapters.common import annotate_raw from topicforge.constants import MAX_SAMPLE_COUNT from topicforge.models import ( BagAnalysis, @@ -275,8 +271,3 @@ def _decode_bag_message(reader: Any, connection: Any, raw: bytes) -> dict[str, A if isinstance(decoded, dict): decoded.setdefault("_msgtype", getattr(connection, "msgtype", "")) return decoded - - -# Unused imports kept for forward-compat (planned use in v0.4.0 Phase 3 -# patches that surface participant metadata from MCAP channel records). -_ = (annotate_full, annotate_partial) diff --git a/src/topicforge/tools/handlers.py b/src/topicforge/tools/handlers.py index e106218..fb41cea 100644 --- a/src/topicforge/tools/handlers.py +++ b/src/topicforge/tools/handlers.py @@ -278,11 +278,14 @@ def detect_qos_mismatches( "payloads — each sample's payload may carry " "`_decode_status` (`full`/`partial`/`raw`), `_decode_note` " "(short diagnostic when not `full`), and `_raw_bytes_hex` " - "(serialized bytes preview when the binding could not " - "resolve the IDL/XTypes dynamically). Cyclone uses " - "`cyclonedds.dynamic` ; Fast DDS 2.6.x falls back to raw " - "bytes more often because its dynamic XTypes binding is " - "partial. **Read-only by architecture** — the " + "(serialized-bytes preview). **Caveat**: on the current " + "user-topic `raw` path this preview is empty — a `raw` status " + "means 'topic present on the bus but not decoded', not 'here " + "are the bytes to re-decode' (capturing the on-wire CDR bytes " + "is roadmapped). Cyclone uses `cyclonedds.dynamic` for " + "full/partial decode ; Fast DDS 2.6.x lands on the raw path " + "more often because its dynamic XTypes binding is partial. " + "**Read-only by architecture** — the " "`MiddlewareAdapter` protocol does not expose a write " "method. **Raises an MCP error** when no DDS module is " "active OR when the topic is not announced on the bus." diff --git a/tests/integration/test_scenarios_schema.py b/tests/integration/test_scenarios_schema.py index 373c08a..46a428c 100644 --- a/tests/integration/test_scenarios_schema.py +++ b/tests/integration/test_scenarios_schema.py @@ -9,7 +9,7 @@ from pathlib import Path -# Tools we ship as of v0.4.0 Phase 2 — every scenario assertion must +# All 11 tools we ship as of v0.4.0 Phase 3 — every scenario assertion must # target one of these. Mirrors `tests/test_tools_integration.py::MVP_TOOLS`. _KNOWN_TOOLS: set[str] = { "health_check", @@ -22,6 +22,7 @@ "peek_dds_samples", "participant_events", "topic_metrics", + "peek_bag_samples", } # Vendor tags accepted in scenario `required_vendors` lists. diff --git a/tests/test_bag_service.py b/tests/test_bag_service.py index 873b1b4..972da6e 100644 --- a/tests/test_bag_service.py +++ b/tests/test_bag_service.py @@ -161,10 +161,13 @@ def test_bag_service_analyze_returns_enriched_bag_analysis_db3( typestore = get_typestore(Stores.LATEST) string_msgtype = "std_msgs/msg/String" - with Writer(bag_path) as writer: + # rosbags >= 0.10 made `version` a required keyword-only argument on + # Writer (rosbag2 metadata schema version). Use VERSION_LATEST so the + # test tracks the newest schema the installed rosbags supports. + with Writer(bag_path, version=Writer.VERSION_LATEST) as writer: conn = writer.add_connection("/test_topic", string_msgtype, typestore=typestore) for i in range(5): - msg = typestore.types[string_msgtype.replace("/", "__")](data=f"hello-{i}") + msg = typestore.types[string_msgtype](data=f"hello-{i}") writer.write(conn, i * 100_000_000, typestore.serialize_cdr(msg, string_msgtype)) svc = BagService() diff --git a/tests/test_cdr_decoder.py b/tests/test_cdr_decoder.py index e1ca856..e1f73ae 100644 --- a/tests/test_cdr_decoder.py +++ b/tests/test_cdr_decoder.py @@ -212,3 +212,37 @@ def test_extract_publish_ns_header_stamp() -> None: def test_extract_publish_ns_none_when_unavailable() -> None: assert extract_publish_ns_from_payload({}) is None assert extract_publish_ns_from_payload({"header": {"frame_id": "x"}}) is None + + +def test_iter_field_names_string_slots_not_exploded() -> None: + # Audit C2: `__slots__ = "value"` (a bare string) is legal Python; list() + # on it would explode into ['v','a','l','u','e']. It must be treated as a + # single field name. + class OneSlot: + __slots__ = "value" + + def __init__(self) -> None: + self.value = 42 + + assert iter_field_names(OneSlot()) == ["value"] + + +def test_decode_string_slots_object_decodes_single_field() -> None: + class OneSlot: + __slots__ = "value" + + def __init__(self) -> None: + self.value = 42 + + assert decode_field_value(OneSlot()) == {"value": 42} + + +def test_decode_field_value_caps_recursion_depth() -> None: + # Audit M6: a pathologically deep list nest collapses to repr() at the cap + # instead of raising RecursionError. + nested: Any = 0 + for _ in range(100): + nested = [nested] + # Must not raise; the deep interior is repr()'d once the cap is hit. + result = decode_field_value(nested) + assert isinstance(result, list) diff --git a/tests/test_dds_helpers.py b/tests/test_dds_helpers.py index 3ec74df..26ba801 100644 --- a/tests/test_dds_helpers.py +++ b/tests/test_dds_helpers.py @@ -5,12 +5,29 @@ from __future__ import annotations +import pytest + +from topicforge.adapters.base import AdapterError from topicforge.adapters.common import ( DDS_ONLY_ERROR_MSG, canonicalize_vendor_id, format_guid, + validate_domain_id, ) + +def test_validate_domain_id_accepts_range_bounds() -> None: + validate_domain_id(0) + validate_domain_id(232) # no raise + + +def test_validate_domain_id_rejects_out_of_range() -> None: + with pytest.raises(AdapterError, match="domain_id"): + validate_domain_id(-1) + with pytest.raises(AdapterError, match="domain_id"): + validate_domain_id(233) + + # --------------------------------------------------------------------------- # canonicalize_vendor_id — OMG vendor_id mapping # --------------------------------------------------------------------------- @@ -137,3 +154,27 @@ def test_dds_only_error_msg_lists_affected_tools() -> None: blocked.""" for tool in ("list_topics", "get_topic_info", "sample_messages", "analyze_bag"): assert tool in DDS_ONLY_ERROR_MSG, f"missing tool name in error message: {tool!r}" + + +def test_every_canonical_vendor_tag_is_valid_participant_literal() -> None: + """Audit P2-3: pin that every tag `canonicalize_vendor_id` can produce + (the `_VENDOR_ID_MAP` values) is accepted by the `ParticipantInfo.vendor` + Literal. Otherwise an adapter emitting a mapped-but-unlisted tag would + raise a ValidationError at output-construction time. This test fails if + the vendor map and the schema Literal ever drift apart.""" + from topicforge.adapters.common.dds_helpers import _VENDOR_ID_MAP + from topicforge.models import ParticipantEvent, ParticipantInfo + + tags = set(_VENDOR_ID_MAP.values()) | {"cyclone", "fast", "rti", "mock", "unknown"} + for tag in tags: + info = ParticipantInfo(guid="g", vendor=tag, domain_id=0, mode_effective="mock") + assert info.vendor == tag + event = ParticipantEvent( + guid="g", + event_type="discovered", + vendor=tag, + timestamp_ns=0, + domain_id=0, + mode_effective="mock", + ) + assert event.vendor == tag diff --git a/tests/test_dds_introspection.py b/tests/test_dds_introspection.py new file mode 100644 index 0000000..2337eb0 --- /dev/null +++ b/tests/test_dds_introspection.py @@ -0,0 +1,215 @@ +"""Tests for the binding-free DDS discovery-sample introspection helpers. + +Extracted from the Cyclone and Fast adapters (Lot 0, audit 2026-07-08) so +the `getattr`-with-fallback field extraction is testable without the +`cyclonedds` / `fastdds` bindings. The helpers stay vendor-qualified +because the two vendors expose subtly different sample shapes; the tests +pin those documented differences (notably: Fast reads only `topic_name` +while Cyclone also falls back to `topic`). + +Synthetic duck-typed objects only. +""" + +from __future__ import annotations + +import pytest + +from topicforge.adapters.common.dds_introspection import ( + cyclone_extract_guid, + cyclone_extract_hostname, + cyclone_extract_topic_name, + cyclone_extract_vendor_id, + fast_extract_guid, + fast_extract_hostname, + fast_extract_topic_name, + fast_extract_vendor_id, + is_removal, +) + + +class _Obj: + """Minimal attribute bag for building duck-typed discovery samples.""" + + def __init__(self, **attrs: object) -> None: + for key, value in attrs.items(): + setattr(self, key, value) + + +# ------------------------------- is_removal -------------------------------- + + +@pytest.mark.parametrize( + ("status", "expected"), + [ + (None, False), + ("ALIVE", False), + ("REMOVED_PARTICIPANT", True), + ("participant DISPOSED", True), + ("DROPPED", True), + (0, False), + ], +) +def test_is_removal(status: object, expected: bool): + assert is_removal(status) is expected + + +def test_is_removal_enum_like_object(): + class _Status: + def __str__(self) -> str: + return "REMOVED_DURABLE_READER" + + assert is_removal(_Status()) is True + + +# ---------------------------- Cyclone extractors --------------------------- + + +def test_cyclone_extract_guid_bytes(): + assert cyclone_extract_guid(_Obj(key=b"\x01" * 16)) == b"\x01" * 16 + + +def test_cyclone_extract_guid_inner_value(): + assert cyclone_extract_guid(_Obj(key=_Obj(value=b"\x02" * 16))) == b"\x02" * 16 + + +def test_cyclone_extract_guid_missing_returns_none(): + assert cyclone_extract_guid(_Obj()) is None + + +def test_cyclone_extract_vendor_id_bytes(): + assert cyclone_extract_vendor_id(_Obj(vendor_id=b"\x01\x16")) == (1, 22) + + +def test_cyclone_extract_vendor_id_tuple(): + assert cyclone_extract_vendor_id(_Obj(vendor_id=(1, 5))) == (1, 5) + + +def test_cyclone_extract_vendor_id_missing_returns_none(): + assert cyclone_extract_vendor_id(_Obj()) is None + + +def test_cyclone_extract_hostname_str_and_bytes(): + assert cyclone_extract_hostname(_Obj(hostname="robot1")) == "robot1" + assert cyclone_extract_hostname(_Obj(hostname=b"robot2")) == "robot2" + + +def test_cyclone_extract_hostname_missing_returns_none(): + assert cyclone_extract_hostname(_Obj()) is None + + +def test_cyclone_extract_topic_name_primary_and_fallback(): + assert cyclone_extract_topic_name(_Obj(topic_name="/scan")) == "/scan" + # Cyclone falls back to `topic` when `topic_name` is absent. + assert cyclone_extract_topic_name(_Obj(topic="/tf")) == "/tf" + + +def test_cyclone_extract_topic_name_missing_returns_none(): + assert cyclone_extract_topic_name(_Obj()) is None + + +# ------------------------------ Fast extractors ---------------------------- + + +def test_fast_extract_guid_bytes(): + assert fast_extract_guid(_Obj(guid=b"\x03" * 16)) == b"\x03" * 16 + + +def test_fast_extract_guid_from_int_sequence(): + assert fast_extract_guid(_Obj(guid=(1, 2, 3))) == bytes([1, 2, 3]) + + +def test_fast_extract_guid_inner_data(): + assert fast_extract_guid(_Obj(key=_Obj(data=b"\x04" * 16))) == b"\x04" * 16 + + +def test_fast_extract_guid_missing_returns_none(): + assert fast_extract_guid(_Obj()) is None + + +def test_fast_extract_vendor_id_from_nested_info(): + assert fast_extract_vendor_id(_Obj(info=_Obj(vendor_id=b"\x01\x05"))) == (1, 5) + + +def test_fast_extract_vendor_id_tuple(): + assert fast_extract_vendor_id(_Obj(vendor_id=(1, 22))) == (1, 22) + + +def test_fast_extract_vendor_id_missing_returns_none(): + assert fast_extract_vendor_id(_Obj()) is None + + +def test_fast_extract_hostname_reads_name_attr(): + # Fast checks `name` (Cyclone does not) — pin the difference. + assert fast_extract_hostname(_Obj(name="node_x")) == "node_x" + + +def test_fast_extract_topic_name_only_topic_name_attr(): + assert fast_extract_topic_name(_Obj(topic_name="/img")) == "/img" + # Unlike Cyclone, Fast does NOT fall back to `topic`. + assert fast_extract_topic_name(_Obj(topic="/img")) is None + + +def test_common_reexports_are_importable(): + # The adapters import these via the package `__init__`, not the submodule. + from topicforge.adapters import common + + assert callable(common.cyclone_extract_guid) + assert callable(common.fast_qos_to_profile) + assert callable(common.is_removal) + + +# ------------------ defensive fallback branches (binding-shape variance) ---- +# These pin the getattr-fallback paths that exist precisely to absorb +# cross-binding-version shape differences — the branches the audit flagged as +# the silent-failure risk if they ever regress. + + +def test_cyclone_extract_guid_skips_absent_attrs_then_finds_guid(): + # key / participant_key absent → loop continues to the `guid` attr. + assert cyclone_extract_guid(_Obj(guid=b"\x07" * 16)) == b"\x07" * 16 + + +def test_cyclone_extract_guid_non_bytes_without_value_returns_none(): + assert cyclone_extract_guid(_Obj(key=12345)) is None + + +def test_cyclone_extract_vendor_id_from_vendor_attr(): + # Falls back from `vendor_id` to the `vendor` attribute. + assert cyclone_extract_vendor_id(_Obj(vendor=b"\x01\x16")) == (1, 22) + + +def test_cyclone_extract_vendor_id_inner_vendorId_attr(): + assert cyclone_extract_vendor_id(_Obj(vendor_id=_Obj(vendorId=(1, 5)))) == (1, 5) + + +def test_cyclone_extract_hostname_from_user_data_bytes(): + assert cyclone_extract_hostname(_Obj(user_data=b"ud-host")) == "ud-host" + + +def test_fast_extract_guid_inner_value_bytes(): + assert fast_extract_guid(_Obj(key=_Obj(value=b"\x08" * 16))) == b"\x08" * 16 + + +def test_fast_extract_guid_inner_guidprefix_sequence(): + assert fast_extract_guid(_Obj(guid=_Obj(guidPrefix=[1, 2, 3]))) == bytes([1, 2, 3]) + + +def test_fast_extract_guid_non_coercible_sequence_returns_none(): + # A sequence of non-ints can't be coerced to bytes → None, no raise. + assert fast_extract_guid(_Obj(guid=["x", "y"])) is None + + +def test_fast_extract_vendor_id_inner_vendor_id_attr(): + assert fast_extract_vendor_id(_Obj(vendor_id=_Obj(vendor_id=(1, 5)))) == (1, 5) + + +def test_fast_extract_hostname_bytes(): + assert fast_extract_hostname(_Obj(hostname=b"fasthost")) == "fasthost" + + +def test_fast_extract_hostname_missing_returns_none(): + assert fast_extract_hostname(_Obj()) is None + + +def test_fast_extract_topic_name_missing_returns_none(): + assert fast_extract_topic_name(_Obj()) is None diff --git a/tests/test_dds_qos_normalization.py b/tests/test_dds_qos_normalization.py new file mode 100644 index 0000000..d5fabb3 --- /dev/null +++ b/tests/test_dds_qos_normalization.py @@ -0,0 +1,286 @@ +"""Tests for the vendor QoS → canonical QosProfile normalizers. + +These were extracted from the Cyclone and Fast adapters (Lot 0, audit +2026-07-08) precisely so they can be tested WITHOUT the `cyclonedds` / +`fastdds` bindings installed. Before the extraction the entire QoS +normalization path — the feeder of `detect_qos_mismatches`, the flagship +DDS diagnostic — was unreachable by the suite, so a renamed policy key +would silently make every QoS profile resolve to `None` (→ no mismatch +ever reported) with the suite still green. The `*_returns_none` cases +below pin exactly that failure mode. + +Synthetic duck-typed objects only — no DDS middleware required. +""" + +from __future__ import annotations + +import pytest + +from topicforge.adapters.common import ( + cyclone_qos_to_profile, + detect_mismatches, + fast_qos_to_profile, +) + +# --------------------------------------------------------------------------- +# Cyclone: policies are objects whose class NAME is read (e.g. "Reliable"). +# The synthetic classes below reproduce that shape. +# --------------------------------------------------------------------------- + + +class Reliable: + pass + + +class BestEffort: + pass + + +class Volatile: + pass + + +class TransientLocal: + pass + + +class KeepAll: + pass + + +class KeepLast: + def __init__(self, depth: int = 10) -> None: + self.depth = depth + + +class _Duration: + def __init__(self, ns: int) -> None: + self._ns = ns + + def to_nanoseconds(self) -> int: + return self._ns + + +class Deadline: + def __init__(self, ns: int) -> None: + self.duration = _Duration(ns) + + +class _CycloneSample: + def __init__(self, qos: object) -> None: + self.qos = qos + + +def test_cyclone_full_profile(): + sample = _CycloneSample([Reliable(), TransientLocal(), KeepLast(depth=5), Deadline(1_000_000)]) + profile = cyclone_qos_to_profile(sample) + assert profile is not None + assert profile.reliability == "RELIABLE" + assert profile.durability == "TRANSIENT_LOCAL" + assert profile.history == "KEEP_LAST" + assert profile.history_depth == 5 + assert profile.deadline_ns == 1_000_000 + + +def test_cyclone_no_deadline_still_builds_profile(): + profile = cyclone_qos_to_profile(_CycloneSample([BestEffort(), Volatile(), KeepAll()])) + assert profile is not None + assert profile.reliability == "BEST_EFFORT" + assert profile.durability == "VOLATILE" + assert profile.history == "KEEP_ALL" + assert profile.deadline_ns is None + + +def test_cyclone_missing_reliability_returns_none(): + assert cyclone_qos_to_profile(_CycloneSample([Volatile(), KeepLast()])) is None + + +def test_cyclone_no_qos_attr_returns_none(): + assert cyclone_qos_to_profile(object()) is None + + +def test_cyclone_non_iterable_qos_returns_none(): + # `for policy in qos` raises TypeError → defensively swallowed → None. + assert cyclone_qos_to_profile(_CycloneSample(qos=42)) is None + + +def test_cyclone_renamed_policy_class_returns_none(): + # Regression guard: a binding that renames "Reliable" → "Reliability" + # must make the profile resolve to None (no false mismatch), NOT + # silently pass. This is the exact failure mode the audit flagged as + # previously untestable. + class Reliability: # wrong name — not the spec-canonical "Reliable" + pass + + assert cyclone_qos_to_profile(_CycloneSample([Reliability(), Volatile(), KeepLast()])) is None + + +# --------------------------------------------------------------------------- +# Fast: QoS is a struct with .reliability/.durability/.history/.deadline, +# each exposing an integer `.kind` mapped via binding-derived int→str maps. +# --------------------------------------------------------------------------- + +_REL = {1: "RELIABLE", 0: "BEST_EFFORT"} +_DUR = {0: "VOLATILE", 1: "TRANSIENT_LOCAL", 2: "TRANSIENT", 3: "PERSISTENT"} +_HIST = {0: "KEEP_LAST", 1: "KEEP_ALL"} + + +class _Kind: + def __init__(self, kind: int, depth: int | None = None) -> None: + self.kind = kind + if depth is not None: + self.depth = depth + + +class _Period: + def __init__(self, seconds: int = 0, nanosec: int = 0) -> None: + self.seconds = seconds + self.nanosec = nanosec + + +class _FastDeadline: + def __init__(self, seconds: int = 0, nanosec: int = 0) -> None: + self.period = _Period(seconds, nanosec) + + +class _FastQos: + def __init__( + self, + *, + rel: int, + dur: int, + hist: int, + depth: int = 1, + deadline: _FastDeadline | None = None, + ) -> None: + self.reliability = _Kind(rel) + self.durability = _Kind(dur) + self.history = _Kind(hist, depth) + self.deadline = deadline + + +class _FastSample: + def __init__(self, qos: object) -> None: + self.qos = qos + + +def _fast(sample: object): + return fast_qos_to_profile(sample, reliability_map=_REL, durability_map=_DUR, history_map=_HIST) + + +def test_fast_full_profile(): + qos = _FastQos(rel=1, dur=1, hist=0, depth=7, deadline=_FastDeadline(seconds=1, nanosec=500)) + profile = _fast(_FastSample(qos)) + assert profile is not None + assert profile.reliability == "RELIABLE" + assert profile.durability == "TRANSIENT_LOCAL" + assert profile.history == "KEEP_LAST" + assert profile.history_depth == 7 + assert profile.deadline_ns == 1_000_000_500 + + +def test_fast_no_deadline_still_builds_profile(): + profile = _fast(_FastSample(_FastQos(rel=0, dur=0, hist=1))) + assert profile is not None + assert profile.reliability == "BEST_EFFORT" + assert profile.history == "KEEP_ALL" + assert profile.deadline_ns is None + + +def test_fast_no_qos_returns_none(): + assert _fast(_FastSample(None)) is None + + +def test_fast_unknown_reliability_kind_returns_none(): + # Regression guard: an enum int not in the map (renamed / shifted across + # a binding major version) must resolve the whole profile to None. + assert _fast(_FastSample(_FastQos(rel=99, dur=1, hist=0))) is None + + +# --------------------------------------------------------------------------- +# End-to-end: normalized profiles feed the flagship analyzer correctly. +# --------------------------------------------------------------------------- + + +def test_normalized_profiles_feed_detect_mismatches(): + reader = cyclone_qos_to_profile(_CycloneSample([Reliable(), Volatile(), KeepLast()])) + writer = cyclone_qos_to_profile(_CycloneSample([BestEffort(), Volatile(), KeepLast()])) + assert reader is not None and writer is not None + result = detect_mismatches(reader, writer) + assert result is not None + policies, severity = result + assert "Reliability" in policies + assert severity == "incompatible" + + +@pytest.mark.parametrize("normalizer", ["cyclone", "fast"]) +def test_both_vendors_agree_on_identical_reliable_pair(normalizer: str): + if normalizer == "cyclone": + profile = cyclone_qos_to_profile(_CycloneSample([Reliable(), Volatile(), KeepLast()])) + else: + profile = _fast(_FastSample(_FastQos(rel=1, dur=0, hist=0))) + assert profile is not None + # Identical profile pair is always compatible regardless of vendor path. + assert detect_mismatches(profile, profile) is None + + +# ------------------ deadline / defensive branch coverage -------------------- +# The Deadline policy shape and the Fast period fields vary across binding +# versions; these pin the getattr-fallback branches. + + +def _deadline_policy(*, duration: object = None, deadline: object = None) -> object: + # An instance whose `type(obj).__name__ == "Deadline"` (what the Cyclone + # normalizer keys on), with configurable duration / deadline attributes. + policy = type("Deadline", (), {})() + if duration is not None: + policy.duration = duration + if deadline is not None: + policy.deadline = deadline + return policy + + +def test_cyclone_deadline_duration_as_int(): + sample = _CycloneSample([Reliable(), Volatile(), KeepLast(), _deadline_policy(duration=1234)]) + profile = cyclone_qos_to_profile(sample) + assert profile is not None + assert profile.deadline_ns == 1234 + + +def test_cyclone_deadline_via_deadline_attr_with_to_nanoseconds(): + policy = _deadline_policy(deadline=_Duration(5678)) # `.duration` absent + sample = _CycloneSample([Reliable(), Volatile(), KeepLast(), policy]) + profile = cyclone_qos_to_profile(sample) + assert profile is not None + assert profile.deadline_ns == 5678 + + +def test_fast_qos_missing_reliability_object_returns_none(): + class _Qos: # no reliability / m_reliability attribute at all + durability = _Kind(0) + history = _Kind(0, 1) + deadline = None + + assert _fast(_FastSample(_Qos())) is None + + +def test_fast_deadline_alt_period_field_names(): + # Some bindings expose `.sec` / `.nanoseconds` instead of + # `.seconds` / `.nanosec` — the normalizer falls back to both. + class _Period: + sec = 2 + nanoseconds = 250 + + class _Deadline: + period = _Period() + + qos = _FastQos(rel=1, dur=0, hist=0, deadline=_Deadline()) # type: ignore[arg-type] + profile = _fast(_FastSample(qos)) + assert profile is not None + assert profile.deadline_ns == 2_000_000_250 + + +def test_fast_history_depth_populated(): + profile = _fast(_FastSample(_FastQos(rel=1, dur=0, hist=0, depth=42))) + assert profile is not None + assert profile.history_depth == 42 diff --git a/tests/test_lifecycle_buffer.py b/tests/test_lifecycle_buffer.py index 2125c13..085cd80 100644 --- a/tests/test_lifecycle_buffer.py +++ b/tests/test_lifecycle_buffer.py @@ -8,7 +8,11 @@ import threading -from topicforge.adapters.common.lifecycle import MAX_EVENTS, LifecycleBuffer +from topicforge.adapters.common.lifecycle import ( + MAX_EVENTS, + MAX_PARTICIPANTS, + LifecycleBuffer, +) def test_record_seen_inserts_new_participant() -> None: @@ -216,3 +220,36 @@ def feed(prefix: str) -> None: def test_default_max_events_is_200() -> None: """Pin the constant — it is documented in the tool description.""" assert MAX_EVENTS == 200 + + +def test_default_max_participants_is_4096() -> None: + assert MAX_PARTICIPANTS == 4096 + + +def test_participant_map_bounded_by_max_participants() -> None: + # Audit P1-4: a churny bus (each restart mints a fresh GUID) must not + # grow the participant map without bound. + buf = LifecycleBuffer(max_participants=3) + for i in range(10): + buf.record_seen(guid=f"g{i}", vendor="cyclone", hostname=None, domain_id=0, now_ns=i) + assert len(buf.snapshot_participants()) == 3 + + +def test_eviction_prefers_left_tombstones_over_active() -> None: + buf = LifecycleBuffer(max_participants=3) + for i in range(3): + buf.record_seen(guid=f"g{i}", vendor="cyclone", hostname=None, domain_id=0, now_ns=i) + # Tombstone g1 (status → "left"); dict still holds 3 entries. + buf.record_lost(guid="g1", now_ns=100) + # New arrival at cap → the tombstone is evicted, actives survive. + buf.record_seen(guid="g_new", vendor="cyclone", hostname=None, domain_id=0, now_ns=200) + guids = {p.guid for p in buf.snapshot_participants()} + assert guids == {"g0", "g2", "g_new"} + + +def test_eviction_falls_back_to_oldest_when_all_active() -> None: + buf = LifecycleBuffer(max_participants=2) + for guid, ts in (("a", 1), ("b", 2), ("c", 3)): + buf.record_seen(guid=guid, vendor="cyclone", hostname=None, domain_id=0, now_ns=ts) + # No tombstones → oldest-inserted ("a") evicted. + assert {p.guid for p in buf.snapshot_participants()} == {"b", "c"} diff --git a/tests/test_metrics_buffer.py b/tests/test_metrics_buffer.py index 434d696..f3e4987 100644 --- a/tests/test_metrics_buffer.py +++ b/tests/test_metrics_buffer.py @@ -52,6 +52,18 @@ def test_count_sequence_gaps_dedupes_duplicates() -> None: assert _count_sequence_gaps([0, 0, 1, 2, 2]) == 0 +def test_count_sequence_gaps_skips_wrap_or_reset_discontinuity() -> None: + # Audit C6: a 16-bit wrap (65535→0) or publisher restart is a + # discontinuity, not tens of thousands of lost samples. + assert _count_sequence_gaps([65534, 65535, 0, 1]) == 0 + + +def test_count_sequence_gaps_counts_small_gap_beside_reset() -> None: + # A genuine small gap (missing 2) is still counted even when a large + # reset-sized jump is present in the same writer's stream. + assert _count_sequence_gaps([0, 1, 3, 900_000, 900_001]) == 1 + + def test_percentile_empty_returns_none() -> None: assert _percentile([], 50) is None @@ -173,6 +185,64 @@ def test_sequence_numbers_unavailable_when_all_none() -> None: assert m.sequence_gaps_count == 0 +def test_frequency_uses_n_minus_1_intervals_not_now() -> None: + # Audit C5 (off-by-one + now-based span): 3 samples 1 s apart span + # 2 s over 2 intervals → 1.0 Hz, independent of now_ns. The old code + # divided count by (now - oldest), giving a now-dependent, ~1.5x rate. + buf = MetricsBuffer() + for i in range(3): + buf.record( + topic="/f", + receive_ns=i * 1_000_000_000, + sequence_number=i, + publish_ns=None, + domain_id=0, + ) + m = buf.compute_metrics(topic="/f", window_seconds=60, now_ns=5_000_000_000) + assert m.samples_observed == 3 + assert m.frequency_hz_observed == 1.0 + + +def test_snapshot_same_timestamp_yields_no_frequency() -> None: + # Audit C5: an opportunistic peek surfaces all samples with one shared + # receive_ns. Co-located samples do not define a rate → None (the old + # code reported count / (now - that_instant), a fabricated number). + buf = MetricsBuffer() + for i in range(5): + buf.record( + topic="/snap", + receive_ns=1_000_000_000, + sequence_number=i, + publish_ns=None, + domain_id=0, + ) + m = buf.compute_metrics(topic="/snap", window_seconds=60, now_ns=3_000_000_000) + assert m.samples_observed == 5 + assert m.frequency_hz_observed is None + + +def test_sequence_gaps_grouped_by_writer() -> None: + # Audit C6: two writers with offset counters (100,101 and 0,1) must NOT + # read as an ~99-wide phantom gap. Grouping by writer_guid keeps each + # writer's contiguous run separate → 0 gaps. + buf = MetricsBuffer() + buf.record( + topic="/w", receive_ns=0, sequence_number=100, publish_ns=None, domain_id=0, writer_guid="A" + ) + buf.record( + topic="/w", receive_ns=1, sequence_number=101, publish_ns=None, domain_id=0, writer_guid="A" + ) + buf.record( + topic="/w", receive_ns=2, sequence_number=0, publish_ns=None, domain_id=0, writer_guid="B" + ) + buf.record( + topic="/w", receive_ns=3, sequence_number=1, publish_ns=None, domain_id=0, writer_guid="B" + ) + m = buf.compute_metrics(topic="/w", window_seconds=60, now_ns=1_000_000_000) + assert m.sequence_numbers_available is True + assert m.sequence_gaps_count == 0 + + def test_sequence_gaps_detected_in_window() -> None: buf = MetricsBuffer() # 0,1,2 then gap 3,4 missing then 5,6 → 2 gaps @@ -306,3 +376,18 @@ def feed(topic: str) -> None: assert buf.sample_count("/t1") == 100 assert buf.sample_count("/t2") == 100 + + +def test_topic_map_bounded_by_max_topics() -> None: + # Audit P2-5: the number of distinct topics must not grow unbounded. + buf = MetricsBuffer(max_topics=3) + for i in range(10): + buf.record(topic=f"/t{i}", receive_ns=0, sequence_number=0, publish_ns=None, domain_id=0) + assert len(buf.snapshot_topics()) == 3 + + +def test_topic_eviction_is_oldest_inserted() -> None: + buf = MetricsBuffer(max_topics=2) + for topic in ("/a", "/b", "/c"): + buf.record(topic=topic, receive_ns=0, sequence_number=0, publish_ns=None, domain_id=0) + assert set(buf.snapshot_topics()) == {"/b", "/c"} diff --git a/tests/test_opendds_adapter.py b/tests/test_opendds_adapter.py index 0d21089..b49bd46 100644 --- a/tests/test_opendds_adapter.py +++ b/tests/test_opendds_adapter.py @@ -26,8 +26,10 @@ def test_constructor_succeeds_for_valid_domain() -> None: assert adapter.effective_mode == "live" -def test_is_available_false_without_pyopendds() -> None: - """Pyopendds is not on PyPI ; the probe must report not-available.""" +def test_is_available_always_false_for_stub() -> None: + """Audit S1: a stub must never report available — even if a `pyopendds` + module is importable — because the factory selects on is_available() and + every method here raises. The Dust stub follows the same rule.""" adapter = OpenDdsAdapter(domain_id=0) assert adapter.is_available() is False diff --git a/tests/test_qos_analyzer.py b/tests/test_qos_analyzer.py index 1475b2a..ccae770 100644 --- a/tests/test_qos_analyzer.py +++ b/tests/test_qos_analyzer.py @@ -260,3 +260,24 @@ def test_mixed_severity_keeps_all_offending_policies() -> None: policies, severity = result assert set(policies) == {"Reliability", "Durability", "Deadline", "History"} assert severity == "incompatible" + + +def test_reader_finite_deadline_writer_none_incompatible() -> None: + """Audit P1-3 / C3: a writer offering no deadline = infinite (loosest) + period, which cannot satisfy a reader that requests a finite deadline. + The pre-audit rule skipped this case (both-must-be-non-None) and + returned a false 'compatible'.""" + reader = _profile(deadline_ns=100_000_000) + writer = _profile(deadline_ns=None) + result = detect_mismatches(reader, writer) + assert result is not None + policies, severity = result + assert policies == ["Deadline"] + assert severity == "incompatible" + + +def test_both_deadline_none_compatible() -> None: + """Both infinite → no deadline constraint on either side → compatible.""" + reader = _profile(deadline_ns=None) + writer = _profile(deadline_ns=None) + assert detect_mismatches(reader, writer) is None diff --git a/tests/test_qos_endpoints.py b/tests/test_qos_endpoints.py new file mode 100644 index 0000000..6aafd15 --- /dev/null +++ b/tests/test_qos_endpoints.py @@ -0,0 +1,156 @@ +"""Tests for `common.qos_endpoints.detect_mismatches_across_endpoints`. + +The endpoint-pairing logic extracted from both DDS adapters (Lot 5) — tested +in isolation with synthetic endpoint objects, and once through the real +Cyclone helpers to pin the exact call shape the adapter makes. No binding. +""" + +from __future__ import annotations + +from topicforge.adapters.common import detect_mismatches_across_endpoints +from topicforge.adapters.common.qos_normalize import cyclone_qos_to_profile +from topicforge.models import QosProfile + + +def _profile( + *, + reliability: str = "RELIABLE", + durability: str = "VOLATILE", + history: str = "KEEP_LAST", +) -> QosProfile: + return QosProfile( + reliability=reliability, # type: ignore[arg-type] + durability=durability, # type: ignore[arg-type] + history=history, # type: ignore[arg-type] + history_depth=10, + ) + + +class _Endpoint: + """Minimal duck-typed discovery endpoint.""" + + def __init__(self, topic: str | None, guid: bytes | None, profile: QosProfile | None) -> None: + self._topic = topic + self._guid = guid + self._profile = profile + + +def _topic(e: _Endpoint) -> str | None: + return e._topic + + +def _guid(e: _Endpoint) -> bytes | None: + return e._guid + + +def _qos(e: _Endpoint) -> QosProfile | None: + return e._profile + + +def _detect(subs: list[_Endpoint], pubs: list[_Endpoint], topic: str | None = None): + return detect_mismatches_across_endpoints( + subs=subs, + pubs=pubs, + topic=topic, + qos_to_profile=_qos, + extract_topic_name=_topic, + extract_guid=_guid, + ) + + +def test_reliable_reader_best_effort_writer_reported() -> None: + reader = _Endpoint("/t", b"\x01" * 16, _profile(reliability="RELIABLE")) + writer = _Endpoint("/t", b"\x02" * 16, _profile(reliability="BEST_EFFORT")) + reports = _detect([reader], [writer]) + assert len(reports) == 1 + r = reports[0] + assert r.topic == "/t" + assert "Reliability" in r.incompatible_policies + assert r.severity == "incompatible" + assert r.reader_guid and r.writer_guid + + +def test_compatible_pair_yields_no_report() -> None: + reader = _Endpoint("/t", b"\x01" * 16, _profile()) + writer = _Endpoint("/t", b"\x02" * 16, _profile()) + assert _detect([reader], [writer]) == [] + + +def test_topic_scoping_filters_other_topics() -> None: + subs = [ + _Endpoint("/a", b"\x01" * 16, _profile(reliability="RELIABLE")), + _Endpoint("/b", b"\x03" * 16, _profile(reliability="RELIABLE")), + ] + pubs = [ + _Endpoint("/a", b"\x02" * 16, _profile(reliability="BEST_EFFORT")), + _Endpoint("/b", b"\x04" * 16, _profile(reliability="BEST_EFFORT")), + ] + reports = _detect(subs, pubs, topic="/a") + assert {r.topic for r in reports} == {"/a"} + + +def test_endpoint_with_unresolvable_topic_skipped() -> None: + reader = _Endpoint(None, b"\x01" * 16, _profile(reliability="RELIABLE")) + writer = _Endpoint("/t", b"\x02" * 16, _profile(reliability="BEST_EFFORT")) + # reader has no topic → no pairing possible. + assert _detect([reader], [writer]) == [] + + +def test_endpoint_with_unresolvable_qos_skipped() -> None: + reader = _Endpoint("/t", b"\x01" * 16, None) # qos_to_profile → None + writer = _Endpoint("/t", b"\x02" * 16, _profile(reliability="BEST_EFFORT")) + assert _detect([reader], [writer]) == [] + + +def test_multiple_readers_and_writers_cartesian() -> None: + subs = [ + _Endpoint("/t", b"\x01" * 16, _profile(reliability="RELIABLE")), + _Endpoint("/t", b"\x03" * 16, _profile(reliability="RELIABLE")), + ] + pubs = [ + _Endpoint("/t", b"\x02" * 16, _profile(reliability="BEST_EFFORT")), + _Endpoint("/t", b"\x04" * 16, _profile(reliability="BEST_EFFORT")), + ] + # 2 readers by 2 writers, all incompatible -> 4 reports. + assert len(_detect(subs, pubs)) == 4 + + +def test_end_to_end_with_real_cyclone_helpers() -> None: + # Pin the exact call shape the Cyclone adapter makes: cyclone_qos_to_profile + # + a topic_name attribute + a bytes `key`. + class Reliable: + pass + + class BestEffort: + pass + + class Volatile: + pass + + class KeepLast: + depth = 10 + + class _CycEndpoint: + def __init__(self, topic: str, key: bytes, reliability_cls: type) -> None: + self.topic_name = topic + self.key = key + self.qos = [reliability_cls(), Volatile(), KeepLast()] + + def _tname(e): + return getattr(e, "topic_name", None) + + def _k(e): + return getattr(e, "key", None) + + reader = _CycEndpoint("/scan", b"\x01" * 16, Reliable) + writer = _CycEndpoint("/scan", b"\x02" * 16, BestEffort) + reports = detect_mismatches_across_endpoints( + subs=[reader], + pubs=[writer], + topic=None, + qos_to_profile=cyclone_qos_to_profile, + extract_topic_name=_tname, + extract_guid=_k, + ) + assert len(reports) == 1 + assert "Reliability" in reports[0].incompatible_policies diff --git a/tests/test_tools_integration.py b/tests/test_tools_integration.py index cb52fdb..b5c868b 100644 --- a/tests/test_tools_integration.py +++ b/tests/test_tools_integration.py @@ -10,6 +10,8 @@ import asyncio +import pytest + from topicforge.config import Settings from topicforge.server import build_app @@ -64,6 +66,27 @@ def test_registered_tools_have_descriptions() -> None: assert t.description, f"{t.name} is missing a description" +def test_adapter_error_propagates_as_tool_error() -> None: + """Contract (CLAUDE.md §8, audit test-gap #4): handlers are thin — + `AdapterError` bubbles up to FastMCP, which surfaces it as an MCP-native + error (isError=true) rather than masking it as a successful result. At the + FastMCP `call_tool` layer this manifests as a `ToolError` carrying the + adapter's message. If a handler ever wrapped errors in a custom success + envelope, this would silently pass a normal result instead of raising.""" + from mcp.server.fastmcp.exceptions import ToolError + + app = _mock_app() + with pytest.raises(ToolError, match="Unknown topic"): + asyncio.run(app.call_tool("get_topic_info", {"topic": "/does_not_exist"})) + + +def test_valid_tool_call_returns_result_not_error() -> None: + """Contrast case: a well-formed call returns a result without raising.""" + app = _mock_app() + result = asyncio.run(app.call_tool("health_check", {})) + assert result is not None + + # Map each tool to the title FastMCP derives from its Pydantic return type. # Pinning these prevents a silent regression to `dict[str, Any]` handlers, # which would degrade outputSchema back to `additionalProperties: True`.