diff --git a/src/jarvis_finance/market_data/prices.py b/src/jarvis_finance/market_data/prices.py index 5506e61..784d94b 100644 --- a/src/jarvis_finance/market_data/prices.py +++ b/src/jarvis_finance/market_data/prices.py @@ -6,6 +6,7 @@ from decimal import Decimal, InvalidOperation from sqlite3 import Connection from typing import Protocol from urllib import error, parse, request +import hashlib import json from jarvis_finance.imports.common import stable_id, utc_now @@ -450,6 +451,137 @@ def _corporate_action_status(conn: Connection, *, instrument_id: str, price_date return "none_known" +def _economic_price_payload( + *, + instrument_id: str, + provider: str, + provider_symbol: str | None, + provider_market: str | None, + price_type: str, + close: Decimal | None, + adjusted_close: Decimal | None, + currency: str, + provider_timestamp: str, + quality_status: str, + source_reference: str, +) -> dict[str, str | None]: + return { + "provider": provider.lower(), + "instrument_id": instrument_id, + "provider_symbol": provider_symbol, + "provider_market": provider_market, + "price_type": price_type, + "close": format(close, "f") if close is not None else "", + "adjusted_close": format(adjusted_close, "f") if adjusted_close is not None else None, + "currency": currency.upper(), + "provider_timestamp": provider_timestamp, + "source_reference": source_reference, + "quality_status": quality_status, + } + + +def _normalise_provider_timestamp(value: str) -> str: + """Canonicalise equivalent timestamp spellings without inventing precision.""" + text = value.strip() + if len(text) == 10: + return date.fromisoformat(text).isoformat() + parsed = datetime.fromisoformat(text.replace("Z", "+00:00")) + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + return parsed.astimezone(timezone.utc).isoformat() + + +def _store_economic_price_observation( + conn: Connection, + *, + instrument_id: str, + provider: str, + provider_symbol: str | None, + provider_market: str | None, + price_type: str, + close: Decimal | None, + adjusted_close: Decimal | None, + currency: str, + provider_timestamp: str, + quality_status: str, + source_reference: str | None, + job_reference: str | None, + created_at: str, +) -> str: + # Serialize identity/version allocation across parallel provider workers. + # The caller commits the complete observation + current-price projection atomically. + if not conn.in_transaction: + conn.execute("BEGIN IMMEDIATE") + canonical_provider_timestamp = _normalise_provider_timestamp(provider_timestamp) + origin_reference = source_reference or ":".join( + part for part in (provider.lower(), provider_symbol or "", provider_market or "") if part + ) + payload = _economic_price_payload( + instrument_id=instrument_id, + provider=provider, + provider_symbol=provider_symbol, + provider_market=provider_market, + price_type=price_type, + close=close, + adjusted_close=adjusted_close, + currency=currency, + provider_timestamp=canonical_provider_timestamp, + quality_status=quality_status, + source_reference=origin_reference, + ) + payload_json = json.dumps(payload, sort_keys=True, separators=(",", ":")) + payload_hash = hashlib.sha256(payload_json.encode()).hexdigest() + source_observation_id = stable_id( + "market-source-observation", + provider.lower(), + instrument_id, + canonical_provider_timestamp, + ) + same = conn.execute( + """SELECT observation_id FROM market_price_observations + WHERE source_observation_id=? AND economic_payload_hash=?""", + (source_observation_id, payload_hash), + ).fetchone() + if same: + return str(same["observation_id"]) + predecessor = conn.execute( + """SELECT observation_id,payload_version,economic_payload_json + FROM market_price_observations WHERE source_observation_id=? + ORDER BY payload_version DESC LIMIT 1""", + (source_observation_id,), + ).fetchone() + version = int(predecessor["payload_version"]) + 1 if predecessor else 1 + observation_id = stable_id("market-observation", source_observation_id, payload_hash) + conn.execute( + """INSERT INTO market_price_observations( + observation_id,source_observation_id,payload_version,supersedes_observation_id, + instrument_id,provider,provider_symbol,provider_market,price_type,close,adjusted_close, + currency,provider_timestamp,source_reference,quality_status,economic_payload_json, + economic_payload_hash,created_at,job_reference + ) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", + ( + observation_id, source_observation_id, version, + str(predecessor["observation_id"]) if predecessor else None, + instrument_id, provider, provider_symbol, provider_market, price_type, + payload["close"], payload["adjusted_close"], payload["currency"], canonical_provider_timestamp, + origin_reference, quality_status, payload_json, payload_hash, created_at, job_reference, + ), + ) + if predecessor: + record_audit_event( + conn, + source="market_price_observation_v2", + action="market_price_observation_corrected", + entity_type="market_price_observation", + entity_id=observation_id, + old_values={"supersedes_observation_id": predecessor["observation_id"]}, + new_values={"source_observation_id": source_observation_id, "payload_version": version}, + confirmed=True, + created_by="system", + ) + return observation_id + + def store_market_price( conn: Connection, *, @@ -468,6 +600,8 @@ def store_market_price( price_type: str = "unadjusted_close", run_id: str | None = None, ) -> str: + if not conn.in_transaction: + conn.execute("BEGIN IMMEDIATE") now = utc_now() fetched = fetched_at or now existing = conn.execute( @@ -475,7 +609,26 @@ def store_market_price( (instrument_id, price_date, provider), ).fetchone() corp_status = _corporate_action_status(conn, instrument_id=instrument_id, price_date=price_date, provider=provider, close=close) if quality_status == "fresh" else "not_checked" - market_price_id = stable_id("marketprice", instrument_id, price_date, provider, provider_symbol or "", now) + provider_observed_at = price_timestamp or price_date + _store_economic_price_observation( + conn, + instrument_id=instrument_id, + provider=provider, + provider_symbol=provider_symbol, + provider_market=provider_market, + price_type=price_type, + close=close, + adjusted_close=adjusted_close, + currency=currency, + provider_timestamp=provider_observed_at, + quality_status=quality_status, + source_reference=None, + job_reference=run_id, + created_at=now, + ) + market_price_id = str(existing["market_price_id"]) if existing else stable_id( + "marketprice", instrument_id, price_date, provider, provider_symbol or "", now + ) conn.execute( """ INSERT INTO market_prices( diff --git a/src/jarvis_finance/storage/migrations.py b/src/jarvis_finance/storage/migrations.py index 9f8994b..806bd30 100644 --- a/src/jarvis_finance/storage/migrations.py +++ b/src/jarvis_finance/storage/migrations.py @@ -8,8 +8,8 @@ from sqlite3 import Connection from .schema import INITIAL_SCHEMA_SQL from .postfinance_schema import create_postfinance_ledger_import_v1 -MIGRATION_VERSION = 52 -MIGRATION_NAME = "052_professional_portfolio_cockpit_v1" +MIGRATION_VERSION = 53 +MIGRATION_NAME = "053_asset_refresh_observation_idempotency" INSTRUMENT_OPTIONAL_COLUMNS = { "position_category": "TEXT", @@ -2963,6 +2963,69 @@ def _create_professional_portfolio_cockpit_v1(conn: Connection) -> None: ) +def _create_asset_refresh_observation_v2(conn: Connection) -> None: + """Append-only economic quotes plus richer job quality counters.""" + + conn.executescript( + """ + CREATE TABLE IF NOT EXISTS market_price_observations ( + observation_id TEXT PRIMARY KEY, + source_observation_id TEXT NOT NULL, + payload_version INTEGER NOT NULL CHECK(payload_version > 0), + supersedes_observation_id TEXT REFERENCES market_price_observations(observation_id), + instrument_id TEXT NOT NULL REFERENCES instruments(instrument_id), + provider TEXT NOT NULL, + provider_symbol TEXT, + provider_market TEXT, + price_type TEXT NOT NULL, + close TEXT NOT NULL, + adjusted_close TEXT, + currency TEXT NOT NULL CHECK(length(currency)=3 AND currency=upper(currency)), + provider_timestamp TEXT NOT NULL, + source_reference TEXT, + job_reference TEXT, + quality_status TEXT NOT NULL, + economic_payload_json TEXT NOT NULL CHECK(json_valid(economic_payload_json)), + economic_payload_hash TEXT NOT NULL, + created_at TEXT NOT NULL, + UNIQUE(source_observation_id, payload_version), + UNIQUE(source_observation_id, economic_payload_hash) + ); + CREATE INDEX IF NOT EXISTS idx_market_price_observations_instrument_time + ON market_price_observations(instrument_id,provider_timestamp,created_at); + CREATE TRIGGER IF NOT EXISTS market_price_observations_no_update + BEFORE UPDATE ON market_price_observations + BEGIN SELECT RAISE(ABORT, 'market price observations are immutable'); END; + CREATE TRIGGER IF NOT EXISTS market_price_observations_no_delete + BEFORE DELETE ON market_price_observations + BEGIN SELECT RAISE(ABORT, 'market price observations cannot be deleted'); END; + """ + ) + _add_missing_columns( + conn, + "market_price_observations", + {"job_reference": "TEXT"}, + ) + _add_missing_columns( + conn, + "portfolio_valuation_snapshots", + { + "source_observation_id": "TEXT", + "economic_payload_hash": "TEXT", + }, + ) + _add_missing_columns( + conn, + "asset_price_refresh_sources", + { + "fresh_unchanged_count": "INTEGER NOT NULL DEFAULT 0", + "stale_remaining_count": "INTEGER NOT NULL DEFAULT 0", + "failed_count": "INTEGER NOT NULL DEFAULT 0", + "diagnostics_json": "TEXT NOT NULL DEFAULT '[]'", + }, + ) + + def _apply_compat_migrations(conn: Connection) -> None: _add_missing_instrument_columns(conn) for table, text_columns in TEXT_AFFINITY_COLUMNS.items(): @@ -3005,6 +3068,7 @@ def _apply_compat_migrations(conn: Connection) -> None: _create_current_source_coverage_and_truewealth_activity_v1(conn) _create_crypto_reconciliation_cockpit_v1(conn) _create_professional_portfolio_cockpit_v1(conn) + _create_asset_refresh_observation_v2(conn) def apply_migrations(conn: Connection) -> None: diff --git a/tests/unit/test_asset_price_refresh.py b/tests/unit/test_asset_price_refresh.py index c70b396..79d1f61 100644 --- a/tests/unit/test_asset_price_refresh.py +++ b/tests/unit/test_asset_price_refresh.py @@ -5,9 +5,11 @@ from decimal import Decimal from pathlib import Path from sqlite3 import Connection import threading +from types import SimpleNamespace from typing import Callable from jarvis_finance.market.providers import PriceQuote +from jarvis_finance.fx.rates import upsert_fx_rate from jarvis_finance.services import asset_price_refresh as asset_refresh from jarvis_finance.services.asset_price_refresh import ( asset_price_refresh_status, @@ -149,7 +151,118 @@ def test_concurrent_workers_claim_a_queued_job_exactly_once(tmp_path): assert conn.execute( "SELECT COUNT(*) FROM aggregated_wealth_refresh_snapshots WHERE job_id=?", (queued["job_id"],), - ).fetchone()[0] == 1 + ).fetchone()[0] == 0 + conn.close() + + +def test_all_failed_sources_create_no_false_wealth_snapshot(tmp_path): + path = tmp_path / "finance.sqlite3" + conn = database(path) + queued, db_path = create_asset_price_refresh_job(conn) + conn.close() + + def failed(_conn, _stale_before): + raise RuntimeError("internal provider detail") + + run_asset_price_refresh(db_path, queued["job_id"], runners={source: failed for source in ("equity", "crypto", "fx")}) + conn = connect(path) + status = asset_price_refresh_status(conn, queued["job_id"]) + assert status["status"] == "failed" + assert status["wealth_snapshot_created"] is False + assert status["failed_assets"] == 3 + assert conn.execute("SELECT COUNT(*) FROM aggregated_wealth_refresh_snapshots").fetchone()[0] == 0 + assert "internal provider detail" not in str(status) + conn.close() + + +def test_partial_instrument_failure_preserves_success_and_quality_counts(tmp_path): + path = tmp_path / "finance.sqlite3" + conn = database(path) + queued, db_path = create_asset_price_refresh_job(conn) + conn.close() + + def equity(_conn, _stale_before): + return asset_refresh.SourceRunResult( + stale_candidates=3, + updated_count=2, + stale_remaining_count=1, + failed_count=1, + diagnostics=("provider_error",), + ) + + def current(_conn, _stale_before): + return asset_refresh.SourceRunResult(fresh_unchanged_count=2) + + run_asset_price_refresh( + db_path, + queued["job_id"], + runners={"equity": equity, "crypto": current, "fx": current}, + ) + conn = connect(path) + status = asset_price_refresh_status(conn, queued["job_id"]) + assert status["status"] == "partial" + assert status["successful_assets"] == 2 + assert status["fresh_unchanged_assets"] == 4 + assert status["stale_assets"] == 1 + assert status["failed_assets"] == 1 + assert status["wealth_snapshot_created"] is True + conn.close() + + +def test_unresolved_stale_asset_makes_job_partial(tmp_path): + path = tmp_path / "finance.sqlite3" + conn = database(path) + queued, db_path = create_asset_price_refresh_job(conn) + conn.close() + + def stale(_conn, _stale_before): + return asset_refresh.SourceRunResult( + stale_candidates=1, + fresh_unchanged_count=1, + stale_remaining_count=1, + ) + + def current(_conn, _stale_before): + return asset_refresh.SourceRunResult(fresh_unchanged_count=1) + + run_asset_price_refresh( + db_path, + queued["job_id"], + runners={"equity": stale, "crypto": current, "fx": current}, + ) + conn = connect(path) + status = asset_price_refresh_status(conn, queued["job_id"]) + assert status["status"] == "partial" + assert status["stale_assets"] == 1 + assert status["wealth_snapshot_created"] is False + source = next(item for item in status["sources"] if item["source"] == "equity") + assert source["status"] == "failed" + assert source["error_code"] == "equity_stale_remaining" + conn.close() + + +def test_equity_valuation_warning_marks_source_partial(tmp_path, monkeypatch): + conn = database(tmp_path / "finance.sqlite3") + response = type( + "Response", + (), + { + "total": 2, + "cached": 0, + "updated": 2, + "economic_updated": 2, + "results": [{"status": "fresh"}, {"status": "fresh"}], + "errors": [], + "warnings": ["portfolio_valuation_partial"], + }, + )() + monkeypatch.setattr(asset_refresh, "refresh_equity_quotes_batch", lambda *_args, **_kwargs: response) + + result = asset_refresh._equity_source(conn, "2026-08-26T00:00:00+00:00") + + assert result.updated_count == 2 + assert result.failed_count == 1 + assert result.diagnostics == ("portfolio_valuation_partial",) conn.close() @@ -190,7 +303,7 @@ def test_crypto_refresh_fetches_only_stale_held_asset_ids(tmp_path, monkeypatch) provider_id: PriceQuote( provider_id, currency, - Decimal("123.45"), + Decimal("100") if provider_id == "ethereum" else Decimal("123.45"), provider_timestamp=now.isoformat(), ) for provider_id in ids @@ -210,4 +323,49 @@ def test_crypto_refresh_fetches_only_stale_held_asset_ids(tmp_path, monkeypatch) assert provider.calls == [("bitcoin",)] assert conn.execute("SELECT COUNT(*) FROM crypto_prices WHERE asset_id='eth'").fetchone()[0] == 1 assert conn.execute("SELECT COUNT(*) FROM crypto_prices WHERE asset_id='btc'").fetchone()[0] == 1 + + replay = asset_refresh._crypto_source(conn, "2099-01-01T00:00:00+00:00") + assert replay.updated_count == 0 + assert replay.fresh_unchanged_count == 2 + assert conn.execute("SELECT COUNT(*) FROM crypto_prices WHERE asset_id='btc'").fetchone()[0] == 1 + assert conn.execute("SELECT COUNT(*) FROM crypto_prices WHERE asset_id='eth'").fetchone()[0] == 1 + conn.close() + + +def test_fx_metadata_replay_is_not_an_economic_update(tmp_path, monkeypatch): + conn = database(tmp_path / "finance.sqlite3") + conn.execute( + "INSERT INTO instruments(instrument_id,asset_class,name,currency,is_active,created_at) " + "VALUES('eur-asset','etf','Synthetic EUR','EUR',1,'2026-01-01')" + ) + upsert_fx_rate( + conn, + base_currency="EUR", + quote_currency="CHF", + rate_date="2026-08-27", + rate=Decimal("0.80448"), + provider="mockfx", + rate_type="close", + fetched_at="2026-08-27T10:00:00+00:00", + ) + conn.commit() + + def resolve_same(conn_arg, **_kwargs): + upsert_fx_rate( + conn_arg, + base_currency="EUR", + quote_currency="CHF", + rate_date="2026-08-27", + rate=Decimal("0.80448"), + provider="mockfx", + rate_type="close", + ) + return SimpleNamespace(status="ok") + + monkeypatch.setattr("jarvis_finance.fx.rates.resolve_fx_rate_to_chf", resolve_same) + result = asset_refresh._fx_source(conn, "2099-01-01T00:00:00+00:00") + + assert result.updated_count == 0 + assert result.fresh_unchanged_count == 1 + assert conn.execute("SELECT COUNT(*) FROM fx_rates WHERE base_currency='EUR'").fetchone()[0] == 1 conn.close() diff --git a/tests/unit/test_market_observation_idempotency.py b/tests/unit/test_market_observation_idempotency.py new file mode 100644 index 0000000..4c49b93 --- /dev/null +++ b/tests/unit/test_market_observation_idempotency.py @@ -0,0 +1,207 @@ +from __future__ import annotations + +from decimal import Decimal +from pathlib import Path +import threading + +from jarvis_finance.market_data.prices import store_market_price +from jarvis_finance.services.portfolio_analytics import run_daily_market_valuation +from jarvis_finance.storage.database import connect +from jarvis_finance.storage.migrations import apply_migrations +from test_portfolio_market_analytics_v1 import Fx, database, quotes_for + + +def test_identical_economic_quote_ignores_request_metadata() -> None: + conn = connect(":memory:") + apply_migrations(conn) + conn.execute( + "INSERT INTO instruments(instrument_id,asset_class,name,currency,is_active,created_at) VALUES('i','stock','Synthetic','CHF',1,'2026-01-01')" + ) + first = store_market_price( + conn, + instrument_id="i", + price_date="2026-08-27", + close=Decimal("10.50"), + currency="CHF", + provider="mock", + provider_symbol="SYN", + price_timestamp="2026-08-27T10:00:00+00:00", + fetched_at="2026-08-27T10:00:01+00:00", + run_id="job-a", + ) + second = store_market_price( + conn, + instrument_id="i", + price_date="2026-08-27", + close=Decimal("10.50"), + currency="CHF", + provider="mock", + provider_symbol="SYN", + price_timestamp="2026-08-27T10:00:00Z", + fetched_at="2026-08-27T10:05:00+00:00", + run_id="job-b", + ) + assert second == first + assert conn.execute("SELECT COUNT(*) FROM market_price_observations").fetchone()[0] == 1 + stored = conn.execute( + "SELECT economic_payload_json,job_reference FROM market_price_observations" + ).fetchone() + assert stored["job_reference"] == "job-a" + assert "job-a" not in stored["economic_payload_json"] + assert "job-b" not in stored["economic_payload_json"] + + store_market_price( + conn, + instrument_id="i", + price_date="2026-08-27", + close=Decimal("10.50"), + currency="CHF", + provider="mock", + provider_symbol="SYN-CORRECTED", + price_timestamp="2026-08-27T10:00:00+00:00", + run_id="job-c", + ) + versions = conn.execute( + """SELECT payload_version,supersedes_observation_id FROM market_price_observations + ORDER BY payload_version""" + ).fetchall() + assert [row["payload_version"] for row in versions] == [1, 2] + assert versions[1]["supersedes_observation_id"] is not None + + +def test_two_same_day_provider_times_and_correction_are_append_only() -> None: + conn = connect(":memory:") + apply_migrations(conn) + conn.execute( + "INSERT INTO instruments(instrument_id,asset_class,name,currency,is_active,created_at) VALUES('i','stock','Synthetic','CHF',1,'2026-01-01')" + ) + for timestamp, close in ( + ("2026-08-27T10:00:00+00:00", "10"), + ("2026-08-27T11:00:00+00:00", "11"), + ("2026-08-27T11:00:00+00:00", "11.1"), + ): + store_market_price( + conn, + instrument_id="i", + price_date="2026-08-27", + close=Decimal(close), + currency="CHF", + provider="mock", + provider_symbol="SYN", + price_timestamp=timestamp, + ) + rows = conn.execute( + "SELECT observation_id,source_observation_id,payload_version,supersedes_observation_id,close FROM market_price_observations ORDER BY created_at,observation_id" + ).fetchall() + assert len(rows) == 3 + assert rows[0]["source_observation_id"] != rows[1]["source_observation_id"] + assert rows[1]["source_observation_id"] == rows[2]["source_observation_id"] + assert (rows[1]["payload_version"], rows[2]["payload_version"]) == (1, 2) + assert rows[2]["supersedes_observation_id"] == rows[1]["observation_id"] + audit = conn.execute( + "SELECT action,old_values_json,new_values_json FROM audit_log WHERE action='market_price_observation_corrected'" + ).fetchone() + assert audit is not None + assert rows[1]["observation_id"] in str(audit["old_values_json"]) + assert [row["close"] for row in rows] == ["10", "11", "11.1"] + + +def test_partial_daily_run_accepts_new_economic_payload_as_new_version(tmp_path: Path) -> None: + conn = database() + first_provider = quotes_for("2026-07-01", missing={"BENCH.S"}) + first = run_daily_market_valuation( + conn, + as_of="2026-07-01", + price_providers={"mock": first_provider}, + fx_provider=Fx({("EUR", "2026-07-01"): "0.95", ("USD", "2026-07-01"): "0.80"}), + lock_path=tmp_path / "job.lock", + ) + assert first.status == "partial" + store_market_price( + conn, + instrument_id="eur", + price_date="2026-07-01", + close=Decimal("51"), + currency="EUR", + provider="mock", + provider_symbol="EUR.S", + provider_market="SIX", + price_timestamp="2026-07-01T21:00:00+00:00", + run_id="provider-correction", + ) + corrected = quotes_for("2026-07-01") + second = run_daily_market_valuation( + conn, + as_of="2026-07-01", + price_providers={"mock": corrected}, + fx_provider=Fx({("EUR", "2026-07-01"): "0.95", ("USD", "2026-07-01"): "0.80"}), + lock_path=tmp_path / "job.lock", + ) + assert second.run_id == first.run_id + assert second.status == "complete" + versions = conn.execute( + "SELECT snapshot_version,supersedes_snapshot_id,value_original FROM portfolio_valuation_snapshots WHERE scope_kind='instrument' AND scope_id='eur' ORDER BY snapshot_version" + ).fetchall() + assert len(versions) == 2 + assert versions[1]["supersedes_snapshot_id"] is not None + assert [row["value_original"] for row in versions] == ["500", "510"] + + +def test_parallel_corrections_allocate_distinct_append_only_versions(tmp_path: Path) -> None: + path = tmp_path / "finance.sqlite3" + conn = connect(path) + apply_migrations(conn) + conn.execute( + "INSERT INTO instruments(instrument_id,asset_class,name,currency,is_active,created_at) " + "VALUES('i','stock','Synthetic','CHF',1,'2026-01-01')" + ) + store_market_price( + conn, + instrument_id="i", + price_date="2026-08-27", + close=Decimal("10"), + currency="CHF", + provider="mock", + provider_symbol="SYN", + price_timestamp="2026-08-27T10:00:00+00:00", + ) + conn.close() + barrier = threading.Barrier(2) + errors: list[Exception] = [] + returned_ids: list[str] = [] + + def correct(value: str) -> None: + worker = connect(path) + try: + barrier.wait(timeout=5) + returned_ids.append(store_market_price( + worker, + instrument_id="i", + price_date="2026-08-27", + close=Decimal(value), + currency="CHF", + provider="mock", + provider_symbol="SYN", + price_timestamp="2026-08-27T10:00:00Z", + )) + except Exception as exc: + errors.append(exc) + finally: + worker.close() + + threads = [threading.Thread(target=correct, args=(value,)) for value in ("11", "12")] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=10) + + assert errors == [] + assert len(set(returned_ids)) == 1 + conn = connect(path) + rows = conn.execute( + """SELECT payload_version,supersedes_observation_id + FROM market_price_observations ORDER BY payload_version""" + ).fetchall() + assert [row["payload_version"] for row in rows] == [1, 2, 3] + assert all(row["supersedes_observation_id"] for row in rows[1:]) + conn.close() __HERMES_CWD_8d46a20096ed__/home/agent/.hermes/worktrees/FinanceManager-sprint23.1__HERMES_CWD_8d46a20096ed__