Implement bounded release runtime and durable native health sampler

Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a126e1-76f8-7643-8cf3-e93336539ade

src/activity_core/api.py

@@ -382,3 +382,10 @@ async def approve_event_type(type_id: str) -> dict[str, str]:
                 {"tid": type_id},
             )
     return {"type_id": type_id, "status": "active", "message": "approved"}
+
+
+@app.get("/release/health")
+async def release_health() -> dict[str, Any]:
+    """Fixed read-only release invariants; no caller-defined operation or target."""
+    from activity_core.release_health import probe
+    return await probe(_get_db(), _get_temporal())

src/activity_core/release_health.py

@@ -0,0 +1,119 @@
+"""Fixed read-only release invariants; no trigger, schedule mutation or credential API."""
+from __future__ import annotations
+
+import asyncio
+from datetime import datetime, timezone, timedelta
+from uuid import NAMESPACE_URL, uuid5, UUID
+
+from sqlalchemy import select
+from temporalio.client import ScheduleActionStartWorkflow, ScheduleOverlapPolicy
+
+from .orm import ActivityDefinition
+
+NAMESPACE = uuid5(NAMESPACE_URL, 'activity-core:activity-definition')
+FRONTEND = {
+    str(uuid5(NAMESPACE, f'frontend-patterns-{cadence}')): (cadence, cron)
+    for cadence, cron in [('daily', '0 2 * * *'), ('weekly', '0 3 * * 2'), ('monthly', '0 9 1 * *')]
+}
+SINK = {'type': 'state-hub-progress', 'event_type': 'fep_feedback_intake',
+        'author': 'activity-core', 'topic_id': 'fe2aaa78-9c20-4feb-b3d2-4fe0529572a3'}
+
+
+def definition_matches(row):
+    expected = FRONTEND.get(str(row.id))
+    if expected is None:
+        return False
+    cadence, cron = expected
+    instructions = row.instructions_json
+    return (row.enabled is True and row.trigger_type == 'cron'
+            and row.trigger_config.get('cron_expression') == cron
+            and row.trigger_config.get('timezone') == 'Europe/Berlin'
+            and row.trigger_config.get('misfire_policy') == 'catchup_latest'
+            and row.trigger_config.get('catchup_window_seconds') == 86400
+            and row.trigger_config.get('jitter_seconds', 0) == 0
+            and isinstance(instructions, list) and len(instructions) == 1
+            and instructions[0].get('id') == f'frontend-patterns-{cadence}-report'
+            and instructions[0].get('model') == 'deterministic'
+            and instructions[0].get('report_sinks') == [SINK])
+
+
+def expanded(ranges, minimum, maximum):
+    values = set()
+    for item in ranges:
+        if not minimum <= item.start <= item.end <= maximum or item.step < 1:
+            raise ValueError('invalid schedule range')
+        values.update(range(item.start, item.end + 1, item.step))
+    return values
+
+
+def schedule_times_match(spec, cadence, cron):
+    if (spec.intervals or spec.skip or spec.start_at is not None or spec.end_at is not None
+            or spec.jitter not in (None, timedelta(0)) or spec.time_zone_name != 'Europe/Berlin'):
+        return False
+    # Temporal converts cron strings to calendar specs on the server. Accept the
+    # exact equivalent fixed calendar, never merely the source cron text.
+    if list(spec.cron_expressions) == [cron] and not spec.calendars:
+        return True
+    if spec.cron_expressions or len(spec.calendars) != 1:
+        return False
+    calendar = spec.calendars[0]
+    hour = {'daily': 2, 'weekly': 3, 'monthly': 9}[cadence]
+    return (not calendar.year and expanded(calendar.second,0,59) == {0}
+            and expanded(calendar.minute,0,59) == {0} and expanded(calendar.hour,0,23) == {hour}
+            and expanded(calendar.day_of_month,1,31) == ({1} if cadence == 'monthly' else set(range(1,32)))
+            and expanded(calendar.month,1,12) == set(range(1,13))
+            and expanded(calendar.day_of_week,0,6) == ({2} if cadence == 'weekly' else set(range(7))))
+
+
+def schedule_matches(description, definition_id, decoded_args):
+    schedule = description.schedule
+    cadence, cron = FRONTEND[definition_id]
+    action = schedule.action
+    return (schedule.state.paused is False and not schedule.state.limited_actions
+            and schedule_times_match(schedule.spec,cadence,cron)
+            and schedule.policy.overlap == ScheduleOverlapPolicy.BUFFER_ONE
+            and schedule.policy.catchup_window == timedelta(seconds=86400)
+            and isinstance(action, ScheduleActionStartWorkflow)
+            and action.workflow == 'RunActivityWorkflow' and action.task_queue == 'orchestrator-tq'
+            and action.id == f'activity-{definition_id}:${{firstScheduledTime}}'
+            and list(decoded_args) == [definition_id, 'scheduled', None])
+
+
+async def schedule_args(temporal, description):
+    action = description.schedule.action
+    if not isinstance(action, ScheduleActionStartWorkflow):
+        raise ValueError('unknown schedule action')
+    # Actual describe() results contain encoded Payload arguments, not Python
+    # values. Decode using the already-bound client converter, never a request codec.
+    return await temporal.data_converter.decode(action.args) if action._from_raw else list(action.args)
+
+
+async def probe(session_factory, temporal):
+    """Read the exact three definitions and their real Temporal schedules, bounded."""
+    result = {'schema_version': 'activity-core-release-health/v1',
+              'observed_at': datetime.now(timezone.utc).isoformat(),
+              'healthy': False, 'definition_count': 0, 'schedule_count': 0,
+              'definitions_and_report_sinks_match': False, 'schedules_match': False}
+    try:
+        async with asyncio.timeout(15):
+            async with session_factory() as session:
+                rows = list((await session.scalars(select(ActivityDefinition).where(
+                    ActivityDefinition.id.in_([UUID(identity) for identity in FRONTEND])))).all())
+            result['definition_count'] = len(rows)
+            if len(rows) != 3 or {str(r.id) for r in rows} != set(FRONTEND) or not all(definition_matches(r) for r in rows):
+                return result
+            result['definitions_and_report_sinks_match'] = True
+            descriptions = await asyncio.gather(*[
+                temporal.get_schedule_handle(f'activity-schedule-{identity}').describe()
+                for identity in sorted(FRONTEND)
+            ])
+            result['schedule_count'] = len(descriptions)
+            decoded = await asyncio.gather(*[schedule_args(temporal,description) for description in descriptions])
+            result['schedules_match'] = all(schedule_matches(description, identity, args)
+                for identity, description, args in zip(sorted(FRONTEND), descriptions, decoded, strict=True))
+            result['healthy'] = result['schedules_match']
+    except Exception:
+        # Database contents, report payloads and Temporal exceptions never escape.
+        result['error'] = 'release invariant observation unavailable'
+    result['observed_at'] = datetime.now(timezone.utc).isoformat()
+    return result

src/activity_core/release_runtime.py

@@ -0,0 +1,201 @@
+"""Owner-configured entrypoint for the existing dedicated Temporal release worker.
+
+No caller-supplied transport factory, signing key, command or admission endpoint.
+A trusted deployment file is configuration, not proof of native authority/custody.
+The actual principal, key issuers and isolated denial/rollback acceptance remain
+owner admission requirements before this process can be deployed.
+"""
+from __future__ import annotations
+
+import argparse
+import asyncio
+import base64
+import json
+import os
+from pathlib import Path
+import sqlite3
+import stat
+
+from temporalio.client import Client
+from temporalio.exceptions import ApplicationError
+
+from .release_broker import Broker, Receipts, ROLES, canonical
+from .release_dispatch import ReleaseActivities, release_worker
+from .release_sampler import CLUSTER_UID, Probe
+from .release_transport import GitArgoBackend
+
+TEMPORAL_HOST = 'actcore-temporal.activity-core.svc.cluster.local:7233'
+FIELDS = {'schema_version','enabled','principal','cluster_uid','state_directory',
+          'kubeconfig','git_config','receipt_trust'}
+
+
+def trusted_file(path):
+    path = Path(path)
+    status = path.lstat()
+    if (not stat.S_ISREG(status.st_mode) or status.st_uid != os.geteuid()
+            or stat.S_IMODE(status.st_mode) & 0o077):
+        raise ValueError('trusted deployment file must be owned and private')
+    return path
+
+
+class Admission:
+    def __init__(self, path):
+        self.path=trusted_file(path)
+        self.value=json.loads(self.path.read_text())
+        value=self.value
+        if (set(value)!=FIELDS or value['schema_version']!='activity-core-release-runtime/v1'
+                or value['enabled'] is not True or value['principal']!='activity-core-release-broker'
+                or value['cluster_uid']!=CLUSTER_UID):
+            raise ValueError('runtime configuration is disabled or outside admitted scope')
+        self.kubeconfig=trusted_file(value['kubeconfig'])
+        self.git_config=trusted_file(value['git_config'])
+        self.configuration=canonical({k:v for k,v in value.items() if k!='enabled'})
+        # Retain only digests for drift/revocation checks, never credential copies.
+        import hashlib
+        self.custody={str(path):hashlib.sha256(path.read_bytes()).hexdigest()
+                     for path in (self.kubeconfig,self.git_config)}
+        self.root=Path(value['state_directory'])
+        self.root.mkdir(parents=True,exist_ok=True,mode=0o700)
+        status=self.root.lstat()
+        if not stat.S_ISDIR(status.st_mode) or status.st_uid!=os.geteuid() or stat.S_IMODE(status.st_mode)&0o077:
+            raise ValueError('runtime state directory must be owned and private')
+        self.trust={}
+        for key_id,entry in value['receipt_trust'].items():
+            if (set(entry)!={'public_key','principal','roles'} or not entry['principal']
+                    or not entry['roles'] or not set(entry['roles'])<=ROLES):
+                raise ValueError('invalid deployment-owned receipt trust')
+            public=base64.b64decode(entry['public_key'],validate=True)
+            if len(public)!=32:raise ValueError('invalid Ed25519 public key')
+            self.trust[key_id]={'public_key':public,'principal':entry['principal'],'roles':set(entry['roles'])}
+        if set().union(*(entry['roles'] for entry in self.trust.values()))!=ROLES:
+            raise ValueError('all independent receipt roles require trusted issuers')
+
+    def require(self):
+        import hashlib
+        current=json.loads(trusted_file(self.path).read_text())
+        if (current.get('enabled') is not True
+                or canonical({k:v for k,v in current.items() if k!='enabled'})!=self.configuration):
+            raise ValueError('runtime admission disabled or deployment configuration changed')
+        if any(hashlib.sha256(trusted_file(path).read_bytes()).hexdigest()!=digest for path,digest in self.custody.items()):
+            raise ValueError('credential custody files changed; owner re-admission required')
+
+    def environment(self):
+        # Prevent fallback to an unrelated global Git credential helper/kubeconfig.
+        return {'KUBECONFIG':str(self.kubeconfig),'GIT_CONFIG_GLOBAL':str(self.git_config),
+                'GIT_CONFIG_NOSYSTEM':'1','GIT_TERMINAL_PROMPT':'0'}
+
+
+class DurableAudit:
+    """Private durable idempotent local receipt sink; no raw transport payloads."""
+    KEYS={'event_id','event_type','release_id','application','phase','observed_at','candidate','rollback'}
+    def __init__(self,path):
+        self.path=Path(path)
+        with self.connect() as db:
+            db.execute('CREATE TABLE IF NOT EXISTS events (event_id TEXT PRIMARY KEY, payload TEXT NOT NULL)')
+
+    def connect(self):
+        db=sqlite3.connect(self.path,timeout=5)
+        db.execute('PRAGMA synchronous=FULL')
+        return db
+
+    def __call__(self,event):
+        if set(event)!=self.KEYS or event['application']!='activity-core' or event['event_type']!='release_transition':
+            raise ValueError('audit event outside release scope')
+        payload=canonical(event).decode()
+        with self.connect() as db:
+            prior=db.execute('SELECT payload FROM events WHERE event_id=?',(event['event_id'],)).fetchone()
+            if prior and prior[0]!=payload:raise ValueError('audit event identity conflict')
+            db.execute('INSERT OR IGNORE INTO events VALUES (?,?)',(event['event_id'],payload))
+
+
+def verify_runtime_identity(runner):
+    """Check the real configured Kubernetes subject and exact allow/deny boundary."""
+    namespace=json.loads(runner(['kubectl','--request-timeout=20s','get','namespace','kube-system','-o','json'],timeout=25))
+    if namespace.get('metadata',{}).get('uid')!=CLUSTER_UID:
+        raise ValueError('runtime cluster UID unverified')
+    identity=json.loads(runner(['kubectl','--request-timeout=20s','auth','whoami','-o','json'],timeout=25))
+    user=identity.get('status',{}).get('userInfo',{})
+    if (user.get('username')!='system:serviceaccount:activity-core:activity-core-release-broker'
+            or 'system:masters' in user.get('groups',[])):
+        raise ValueError('actual release principal differs from dedicated service account')
+    allowed=[('get','argoproj.io','applications','argocd','activity-core'),
+             ('patch','argoproj.io','applications','argocd','activity-core'),
+             ('get','argoproj.io','applications','argocd','railiance-apps-root'),
+             ('patch','argoproj.io','applications','argocd','railiance-apps-root')]
+    denied=[('get','','secrets','activity-core',''),('get','','secrets','argocd',''),
+            ('patch','argoproj.io','applications','argocd','target-revenue'),
+            ('patch','apps','deployments','activity-core','actcore-worker'),
+            ('create','','pods/exec','activity-core',''),
+            ('create','rbac.authorization.k8s.io','rolebindings','activity-core',''),
+            ('create','batch','jobs','activity-core','')]
+    for expected,checks in [(True,allowed),(False,denied)]:
+        for verb,group,resource,namespace,name in checks:
+            attrs={'verb':verb,'group':group,'resource':resource,'namespace':namespace}
+            if '/' in resource:attrs['resource'],attrs['subresource']=resource.split('/',1)
+            if name:attrs['name']=name
+            body={'apiVersion':'authorization.k8s.io/v1','kind':'SelfSubjectAccessReview',
+                  'spec':{'resourceAttributes':attrs}}
+            result=json.loads(runner(['kubectl','--request-timeout=20s','create',
+                '--raw=/apis/authorization.k8s.io/v1/selfsubjectaccessreviews','-f','-'],
+                input=json.dumps(body),timeout=25))
+            status=result.get('status',{})
+            if status.get('allowed') is not expected or status.get('evaluationError'):
+                raise ValueError('release principal allow/deny boundary unverified')
+    return {'principal_verified':True,'allowed_checks':len(allowed),'denied_checks':len(denied)}
+
+
+class RuntimeActivities(ReleaseActivities):
+    def __init__(self,admission):
+        self.admission=admission
+        broker=Broker(admission.root/'release.sqlite',Receipts(admission.trust),admitted=True)
+        env=admission.environment()
+        from .release_transport import run
+        def bounded_run(argv,**kwargs):
+            admission.require()
+            inherited=kwargs.get('env',os.environ)
+            clean={key:value for key,value in inherited.items()
+                   if not key.startswith(('GIT_','KUBE')) and key!='SSH_AUTH_SOCK'}
+            kwargs['env']={**clean,**env}
+            return run(argv,**kwargs)
+        self.identity_probe=lambda:verify_runtime_identity(bounded_run)
+        self.identity_probe()
+        self.probe=Probe(runner=bounded_run)
+        def factory(plan):
+            admission.require()
+            self.identity_probe()
+            def invariants(target):
+                observed,healthy=self.probe()
+                return observed==target and healthy is True
+            backend=GitArgoBackend(plan,admission.root/'transport',invariants,
+                                  cluster_uid=CLUSTER_UID,admitted=True,runner=bounded_run)
+            return backend
+        super().__init__(broker,factory,DurableAudit(admission.root/'audit.sqlite'))
+
+    def _plan(self,release_id):
+        try:self.admission.require()
+        except Exception:
+            self.broker.admitted=False
+            raise ApplicationError('release runtime admission unavailable',non_retryable=True) from None
+        return super()._plan(release_id)
+
+
+async def serve(path):
+    admission=Admission(path)
+    activities=RuntimeActivities(admission)
+    client=await asyncio.wait_for(Client.connect(TEMPORAL_HOST,namespace='default'),timeout=30)
+    await release_worker(client,activities).run()
+
+
+def main():
+    parser=argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--configuration',type=Path,required=True)
+    args=parser.parse_args()
+    try:asyncio.run(serve(args.configuration))
+    except Exception:
+        print(json.dumps({'schema_version':'activity-core-release-runtime/v1','status':'unverified',
+                          'error':'runtime unavailable; private configuration and transport details suppressed'}))
+        return 1
+    return 0
+
+
+if __name__=='__main__':raise SystemExit(main())

src/activity_core/release_sampler.py

@@ -0,0 +1,136 @@
+"""Executable read-only sampler for the existing durable release HealthObserver.
+
+No signer, identity admission or mutation is provided. Each failed/unavailable
+probe records False, missed intervals reset, and a new ledger starts now.
+"""
+from __future__ import annotations
+
+import argparse
+from datetime import datetime, timezone
+import ipaddress
+import json
+from pathlib import Path
+import time
+import urllib.request
+
+import yaml
+
+from .release_observer import HealthObserver
+from .release_operations import revision, update_child
+from .release_transport import run
+
+CLUSTER_UID = 'a553c742-0115-43d4-99a4-a5ca56fe0786'
+BASELINE = 'a86a14332d76d92a7ed176ea34dca4429902c396'
+DEPLOYMENTS = {'actcore-api', 'actcore-worker', 'actcore-event-router'}
+RESOURCE_SET = {
+    ('ConfigMap', name) for name in ('actcore-runtime-config','actcore-external-activity-definitions','actcore-report-schemas','actcore-ops-service-inventory')
+} | {('Service',name) for name in ('actcore-api','actcore-worker-metrics')} | {('Deployment',name) for name in DEPLOYMENTS}
+
+
+def http_health(url):
+    with urllib.request.urlopen(url, timeout=20) as response:
+        if response.status != 200:
+            raise ValueError('invariant service unavailable')
+        body = response.read(16385)
+    if len(body) > 16384:
+        raise ValueError('invariant response exceeds fixed bound')
+    return json.loads(body)
+
+
+def invariant_matches(value, now):
+    if (not isinstance(value, dict) or value.get('schema_version') != 'activity-core-release-health/v1'
+            or value.get('healthy') is not True or value.get('definitions_and_report_sinks_match') is not True
+            or value.get('schedules_match') is not True or value.get('definition_count') != 3
+            or value.get('schedule_count') != 3 or 'error' in value):
+        return False
+    at = datetime.fromisoformat(value['observed_at'].replace('Z', '+00:00'))
+    # Same estate observer should use synchronized clocks; unknown/future time
+    # is a failed sample, never a manufactured healthy interval.
+    return at.tzinfo is not None and 0 <= (now-at).total_seconds() <= 90
+
+
+class Probe:
+    """Only fixed GETs, a cluster-pinned service origin and public boolean health."""
+    def __init__(self, runner=run, reader=http_health):
+        self.runner, self.reader = runner, reader
+
+    def get(self, *args):
+        return json.loads(self.runner(['kubectl','--request-timeout=20s',*args,'-o','json'],timeout=25))
+
+    def __call__(self):
+        nodes = self.get('get','nodes')
+        if '92.205.62.239' not in [a['address'] for n in nodes['items'] for a in n['status']['addresses']]:
+            raise ValueError('estate identity mismatch')
+        if self.get('get','namespace','kube-system')['metadata']['uid'] != CLUSTER_UID:
+            raise ValueError('cluster UID mismatch')
+        app = self.get('-n','argocd','get','application','activity-core')
+        target = revision(app['spec']['source']['targetRevision'])
+        update_child(yaml.safe_dump(app),target,target)
+        status = app.get('status',{})
+        healthy = (not app.get('operation') and status.get('sync',{}).get('status') == 'Synced'
+            and status.get('sync',{}).get('revision') == target
+            and status.get('health',{}).get('status') == 'Healthy' and not status.get('conditions'))
+        resources = status.get('resources',[])
+        healthy &= (len(resources) == 9 and {(r['kind'],r['name']) for r in resources} == RESOURCE_SET
+                    and all(r.get('namespace') == 'activity-core' and r.get('status') == 'Synced'
+                            and (r.get('group') or '') == ('apps' if r['kind']=='Deployment' else '') for r in resources))
+        for name in sorted(DEPLOYMENTS):
+            deployment = self.get('-n','activity-core','get','deployment',name)
+            desired = deployment['spec'].get('replicas',1);ds=deployment.get('status',{})
+            healthy &= (type(desired) is int and desired>0 and ds.get('observedGeneration') == deployment['metadata']['generation']
+                        and all(ds.get(k,0) == desired for k in ('readyReplicas','updatedReplicas','availableReplicas')))
+        service = self.get('-n','activity-core','get','service','actcore-api')
+        address = ipaddress.ip_address(service['spec']['clusterIP'])
+        if not any(p.get('port') == 8010 for p in service['spec'].get('ports',[])):
+            raise ValueError('fixed invariant API port unavailable')
+        host = f'[{address}]' if address.version == 6 else str(address)
+        value = self.reader(f'http://{host}:8010/release/health')
+        healthy &= invariant_matches(value,datetime.now(timezone.utc))
+        return target,bool(healthy)
+
+
+class Sampler:
+    def __init__(self, database, probe, *, anchor_revision=BASELINE):
+        self.observer=HealthObserver(database,max_gap_seconds=90)
+        self.probe=probe;self.anchor=revision(anchor_revision)
+
+    def sample(self,now=None):
+        explicit_now=now
+        try:
+            target,healthy=self.probe()
+            revision(target)
+            if type(healthy) is not bool:raise ValueError('explicit health required')
+        except Exception:
+            # Use the last durably seen identity only for a failed sample. No
+            # raw error, fabricated timestamp or successful identity is emitted.
+            with self.observer.connect() as db:
+                row=db.execute('SELECT revision FROM observations ORDER BY rowid DESC LIMIT 1').fetchone()
+            target=row[0] if row else self.anchor;healthy=False
+        # Measure after the probe, not before a slow/failed network operation.
+        measured=explicit_now if explicit_now is not None else datetime.now(timezone.utc)
+        since=self.observer.observe(target,healthy,measured)
+        try:attestation=self.observer.attestation(target,measured)
+        except ValueError:attestation=None
+        return {'schema_version':'activity-core-release-sample/v1','revision':target,
+                'observed_at':measured.isoformat(),'healthy':healthy,'healthy_since':since,
+                'continuous_24h_ready':attestation is not None,'attestation':attestation,
+                'native_mutation':False}
+
+
+def main():
+    parser=argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--database',type=Path,required=True)
+    parser.add_argument('--receipt',type=Path,required=True)
+    parser.add_argument('--watch',action='store_true',help='read-only 60s sampler; requires an owner-deployed process')
+    args=parser.parse_args()
+    sampler=Sampler(args.database,Probe())
+    while True:
+        try:result=sampler.sample()
+        except Exception:result={'schema_version':'activity-core-release-sample/v1','healthy':False,'continuous_24h_ready':False,'error':'durable observation unavailable','native_mutation':False}
+        args.receipt.write_text(json.dumps(result,sort_keys=True)+'\n')
+        if not args.watch:
+            print(json.dumps(result,sort_keys=True));return 0 if result.get('healthy') else 1
+        time.sleep(60)
+
+
+if __name__=='__main__':raise SystemExit(main())