From dff56856866b7d2825f9bc3e359962a7457151c4 Mon Sep 17 00:00:00 2001 From: flemming-it Date: Thu, 18 Jun 2026 14:18:10 +0200 Subject: [PATCH] =?UTF-8?q?feat(hub):=20implement=20HubRepository.list()?= =?UTF-8?q?=20=E2=80=94=20flow.completed=20=E2=86=92=20Evaluation?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replaces the Phase-0 placeholder (returns const []) with a real event-log reader. When the app starts in hub mode and the hub is healthy, list() now: 1. Calls HubClient.eventLog(types: ['flow.completed']) over gRPC (up to 200 most recent events). 2. Decodes each event's detail JSON as the Recl∆Im output bag produced by flows/durchstich-gewo-14.yaml — keys: norm, skm, benefit, frust, cohort, duties. 3. Skips events from other flows (detected by absence of norm + skm keys). One hub can serve many flows; only Recl∆Im-shaped runs surface here. 4. Sorts by skmEurPerYear DESC, matching MockRepository's 'most-harmful first' ordering — UI behaves identically across the mock-vs-live switch. Field mapping (tolerant on absence, defaults documented): norm.eli / title / jurabk / paragraph / paragraphs[] -> Norm skm.total_eur_per_year, skm.tier_lowest -> skmEurPerYear, tier benefit.total -> benefitScore frust.composite_frust -> frustScore cohort.count -> affectedCount cohort.source + .tier -> first Source entry duties.duties[] -> List event.eventId -> auditEventId event.timestamp -> norm.standDate byEli() walks the same list once; for small per-norm caches we defer batching to Phase 2. Empty list = honest state: 'hub connected, no Recl∆Im flow runs yet'. Caller decides how to render that (HeatmapPage currently shows an empty grid; a clearer 'chain run flows/'- hint lands in the same pass as the demo-banner reorganisation). Signed-off-by: flemming-it --- app/lib/data/hub_repository.dart | 186 +++++++++++++++++++++++++++---- 1 file changed, 162 insertions(+), 24 deletions(-) diff --git a/app/lib/data/hub_repository.dart b/app/lib/data/hub_repository.dart index 4d15458..9c6e35b 100644 --- a/app/lib/data/hub_repository.dart +++ b/app/lib/data/hub_repository.dart @@ -1,3 +1,5 @@ +import 'dart:convert'; + import 'package:chain_client_sdk/chain_client_sdk.dart' as chain; import 'hub_endpoint.dart'; @@ -6,18 +8,17 @@ import 'repository.dart'; /// Live `EvaluationRepository` backed by a Ch∆In hub over gRPC. /// -/// Phase 0 scope: open a HubClient, run a healthy() probe, and -/// surface the connection state to the UI. The list() / byEli() -/// methods return empty data because no flow on the hub yet -/// produces Evaluation objects in the lawheatmap-specific shape — -/// that pipeline lands in Phase 0 week 1 once a chain run wraps -/// flows/durchstich-gewo-14.yaml output into the Evaluation -/// schema. +/// Reads flow.completed events from the hub's audit log, parses +/// each event's `detail` JSON as a Recl∆Im-shaped output bag +/// (see flows/durchstich-gewo-14.yaml), and converts the bag to +/// an Evaluation the UI can render. Events from non-Recl∆Im +/// flows are silently skipped (detected by absence of the +/// `norm` + `skm` keys). /// -/// Until then [MockRepository] stays the default. Pick this -/// repository explicitly via the Hub-settings page; on first -/// healthy() failure the app falls back to MockRepository so the -/// UI keeps working. +/// Sort order matches MockRepository: highest €/Jahr first. +/// If the hub has zero matching flow.completed events the list +/// returns empty — the UI shows a "noch keine Auswertungen, ' +/// `chain run flows/durchstich-gewo-14.yaml` aufrufen"-Hinweis. class HubRepository implements EvaluationRepository { HubRepository._(this._client, this.settings, this._healthy); @@ -25,10 +26,6 @@ class HubRepository implements EvaluationRepository { final HubSettings settings; final bool _healthy; - /// Try to connect. Returns a repository with `isHealthy=false` - /// if the probe times out or the hub is unreachable; the - /// caller decides whether to use it or fall back to - /// MockRepository. static Future connect(HubSettings settings) async { final endpoint = chain.HubEndpoint( host: settings.host, @@ -48,22 +45,32 @@ class HubRepository implements EvaluationRepository { bool get isHealthy => _healthy; - /// Underlying SDK client — exposed so the hub-status page can - /// list capabilities, query the audit log, and verify the - /// event-chain without leaking gRPC types to other pages. chain.HubClient get client => _client; @override Future> list() async { - // Phase 0 placeholder. The pipeline that turns - // flow.completed events into Evaluation objects lives in the - // surrounding flow (flows/durchstich-gewo-14.yaml) and lands - // in Phase 0 week 1. - return const []; + if (!_healthy) return const []; + final events = await _client.eventLog( + limit: 200, + types: const ['flow.completed'], + ); + final evals = []; + for (final e in events) { + final eval = _parse(e); + if (eval != null) evals.add(eval); + } + evals.sort((a, b) => b.skmEurPerYear.compareTo(a.skmEurPerYear)); + return evals; } @override - Future byEli(String eli) async => null; + Future byEli(String eli) async { + final all = await list(); + for (final e in all) { + if (e.norm.eli == eli) return e; + } + return null; + } @override String get mode => _healthy ? 'hub-live' : 'hub-down'; @@ -71,4 +78,135 @@ class HubRepository implements EvaluationRepository { @override String get hubLabel => '${settings.url}' '${_healthy ? '' : ' — Verbindung gestört'}'; + + /// Decode one flow.completed event into an Evaluation, or + /// return null when the event doesn't look like a Recl∆Im + /// flow. We're permissive about field absence so a partially + /// failed flow run still surfaces what it produced. + Evaluation? _parse(chain.LoggedEvent event) { + Map? bag; + try { + final decoded = jsonDecode(event.detail); + if (decoded is Map) bag = decoded; + } catch (_) { + return null; + } + if (bag == null) return null; + + final normJson = _asMap(bag['norm']); + final skmJson = _asMap(bag['skm']); + if (normJson == null || skmJson == null) return null; + + final norm = Norm( + eli: (normJson['eli'] as String?) ?? 'eli/unknown', + title: (normJson['title'] as String?) ?? '(ohne Titel)', + jurabk: (normJson['jurabk'] as String?) ?? '', + paragraph: (normJson['paragraph'] as String?) ?? '', + standDate: _parseTimestamp(event.timestamp), + sourceUrl: (normJson['source_url'] as String?) ?? '', + sourceSha256: (normJson['source_sha256'] as String?) ?? '', + paragraphs: _parseParagraphs(normJson['paragraphs']), + ); + + final skmEur = _asDouble(skmJson['total_eur_per_year']); + final tierLowest = _parseTier(skmJson['tier_lowest']); + + final benefitJson = _asMap(bag['benefit']); + final benefit = _asDouble(benefitJson?['total']); + + final frustJson = _asMap(bag['frust']); + final frust = _asDouble(frustJson?['composite_frust']); + + final cohortJson = _asMap(bag['cohort']); + final affected = _asInt(cohortJson?['count']); + + return Evaluation( + norm: norm, + skmEurPerYear: skmEur, + benefitScore: benefit, + affectedCount: affected, + frustScore: frust, + tierLowest: tierLowest, + duties: _parseDuties(bag['duties']), + sources: _parseSources(cohortJson), + auditEventId: event.eventId, + notes: 'Live aus Hub-Flow ${event.flowName} ' + '(execution ${event.flowExecution.substring(0, 8.clamp(0, event.flowExecution.length))}).', + ); + } + + static Map? _asMap(dynamic v) => + v is Map ? v : null; + + static double _asDouble(dynamic v) { + if (v is num) return v.toDouble(); + if (v is String) return double.tryParse(v) ?? 0.0; + return 0.0; + } + + static int _asInt(dynamic v) { + if (v is num) return v.toInt(); + if (v is String) return int.tryParse(v) ?? 0; + return 0; + } + + static EvidenceTier _parseTier(dynamic v) { + final s = (v as String?)?.toUpperCase() ?? 'T4'; + return switch (s) { + 'T1' => EvidenceTier.t1, + 'T2' => EvidenceTier.t2, + 'T3' => EvidenceTier.t3, + _ => EvidenceTier.t4, + }; + } + + static DateTime _parseTimestamp(String iso) => + DateTime.tryParse(iso) ?? DateTime.now(); + + static List _parseParagraphs(dynamic v) { + if (v is! List) return const []; + return v.whereType>().map((p) { + return NormParagraph( + number: (p['number'] as String?) ?? '', + text: (p['text'] as String?) ?? '', + ); + }).toList(); + } + + static List _parseDuties(dynamic v) { + final outer = _asMap(v); + final inner = outer?['duties']; + if (inner is! List) return const []; + return inner.whereType>().map((d) { + return Duty( + modality: (d['modality'] as String?) ?? '?', + addressee: (d['addressee'] as String?) ?? '', + action: (d['action'] as String?) ?? '', + frequency: (d['frequency'] as String?) ?? '', + sourceNorm: (d['source_norm'] as String?) ?? '', + authority: d['authority'] as String?, + consequence: d['consequence'] as String?, + ); + }).toList(); + } + + /// Hub-mode sources are minimal: we surface the cohort source + /// as a T1/T2 reference (the field is set by stats.cohort_size) + /// plus an audit-log pointer. Richer source attribution comes + /// when the flow has a dedicated `sources.collect` step. + static List _parseSources(Map? cohort) { + final out = []; + final label = cohort?['source'] as String?; + final tierStr = (cohort?['tier'] as String?)?.toUpperCase() ?? 'T2'; + if (label != null && label.isNotEmpty) { + final tier = switch (tierStr) { + 'T1' => EvidenceTier.t1, + 'T2' => EvidenceTier.t2, + 'T3' => EvidenceTier.t3, + _ => EvidenceTier.t4, + }; + out.add(Source(label: label, url: '', tier: tier)); + } + return out; + } }