busya
bcc69f4bbe
Add versioned metric graph authority, owned browser evidence, paginated and all-tab traversal, deterministic sampling policies, and analyst-facing run inspection. Provision the DEV/PREPROD/PROD Superset, Gitea and million-row ClickHouse lab; retain reproducible lifecycle evidence and explicit incomplete-traversal limits.
86 lines
4.1 KiB
Python
86 lines
4.1 KiB
Python
#!/usr/bin/env python3
|
|
# #region FullFlow.FinanceManualObserver [C:3] [TYPE Module] [SEMANTICS fixture,readonly,manual,baseline]
|
|
# @BRIEF Observe the existing queued manual M01 run without creating a duplicate or changing its source.
|
|
# @INVARIANT The run ID must equal the retained actual manual launch; all operations after login are GET-only.
|
|
import argparse
|
|
from decimal import Decimal
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
import time
|
|
from uuid import UUID
|
|
|
|
from release import private_environment, Blocked
|
|
|
|
# The release driver adds historical full_flow to sys.path; resolve the finance driver by its exact file.
|
|
_spec = importlib.util.spec_from_file_location('finance_manual_observer_driver',Path(__file__).with_name('verify.py'))
|
|
_finance = importlib.util.module_from_spec(_spec)
|
|
_spec.loader.exec_module(_finance)
|
|
Driver = _finance.Driver
|
|
|
|
|
|
# #region FullFlow.FinanceManualObserver.Observe [C:3] [TYPE Function]
|
|
# @POST Genuine same-ID terminal metric/comparison and immutable published pin are verified against the fixture oracle.
|
|
def observe(driver, run_id, timeout):
|
|
identity = str(UUID(run_id))
|
|
driver.require('manual-observer-retained-run',driver.ids.get('manual_run_id')==identity)
|
|
deadline = time.monotonic()+timeout
|
|
state = {}
|
|
while time.monotonic() < deadline:
|
|
state = driver.call('manual-observed-state','GET',f'/api/scenario-runs/{identity}')
|
|
if state.get('status') in {'passed','failed','blocked','inconclusive','cancelled','waiting_human','pending_approval'}:
|
|
break
|
|
time.sleep(5)
|
|
driver.require('manual-observed-terminal-pass',state.get('status')=='passed',observed_status=state.get('status'))
|
|
result = driver.call('manual-observed-result','GET',f'/api/scenario-runs/{identity}/result')
|
|
driver.verify_published_pin('manual-observed-published-pin',result)
|
|
rows = result['snapshot']['steps']
|
|
producer = next(row for row in rows if row['logical_step_id']=='phase-3-M01-execute_metric')
|
|
comparison = next(row for row in rows if row['logical_step_id']=='phase-4-M01-compare_to_baseline')
|
|
payload = comparison['step_outcome']
|
|
values = payload.get('step_outcome',payload)['comparison']
|
|
expected = Decimal(driver.config['expected_value'])
|
|
driver.require('manual-observed-real-producer-comparison',result['status']=='passed'
|
|
and producer['status']=='passed' and bool(producer['artifact_refs']) and comparison['status']=='passed'
|
|
and Decimal(values['actual']['canonical_value'])==expected
|
|
and Decimal(values['expected']['canonical_value'])==expected,comparison=values)
|
|
path = driver.output/'manual-result.json'
|
|
if path.exists():
|
|
driver.require('manual-observed-preserves-result',json.loads(path.read_text())==result)
|
|
else:
|
|
path.write_text(json.dumps(result,indent=2,ensure_ascii=False)+'\n')
|
|
# #endregion FullFlow.FinanceManualObserver.Observe
|
|
|
|
|
|
# #region FullFlow.FinanceManualObserver.Main [C:3] [TYPE Function]
|
|
# @PRE Existing finance report/config/private env; bounded wait and actual retained run identity.
|
|
def main():
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument('--env-file',type=Path,required=True)
|
|
parser.add_argument('--fixture',type=Path,required=True)
|
|
parser.add_argument('--output',type=Path,required=True)
|
|
parser.add_argument('--run-id',required=True)
|
|
parser.add_argument('--timeout',type=int,default=7200)
|
|
args = parser.parse_args()
|
|
if not 1 <= args.timeout <= 21600:
|
|
raise SystemExit('Bounded timeout required')
|
|
private_environment(args.env_file)
|
|
prior = json.loads((args.output/'report.json').read_text())
|
|
driver = Driver(json.loads(args.fixture.read_text()),args.output)
|
|
driver.ids,driver.steps = prior['ids'],prior['steps']
|
|
driver.phase = 'manual-observe-existing'
|
|
try:
|
|
driver.login()
|
|
observe(driver,args.run_id,args.timeout)
|
|
except Blocked:
|
|
return 1
|
|
finally:
|
|
driver.flush()
|
|
driver.client.close()
|
|
return 0
|
|
# #endregion FullFlow.FinanceManualObserver.Main
|
|
|
|
if __name__ == '__main__':
|
|
raise SystemExit(main())
|
|
# #endregion FullFlow.FinanceManualObserver
|