feat: prove database-first Leo on private GCP copy

This commit is contained in:
twentyOne2x 2026-07-14 12:32:26 +02:00
parent 6ecc4ac540
commit 9207e1b100
19 changed files with 3511 additions and 178 deletions

View file

@ -13,20 +13,19 @@ real GCP Leo read/reasoning path without Telegram sends or DB mutation.
## Operator Paths ## Operator Paths
The direct alias is passwordless but firewall-source-dependent: The direct alias is passwordless and was live-verified on 2026-07-14:
```bash ```bash
ssh teleo-gcp-staging ssh teleo-gcp-staging
``` ```
It uses `/Users/user/.ssh/google_compute_engine` for It uses `/Users/user/.ssh/google_compute_engine` for the configured operator,
`billy_livingip_xyz@34.65.143.148`, disables password and keyboard-interactive disables password and keyboard-interactive authentication, and does not store a
authentication, and does not store a Google password. It works only while the Google password. `sudo -n` also works. The route remains
Mac's public egress matches `teleo-prod-allow-ssh-current-ip`. On 2026-07-12 the firewall-source-dependent, so verify `ssh -o BatchMode=yes teleo-gcp-staging true` before a
Mac returned to `99.35.221.133` while the rule still allowed long run instead of assuming retained access is current.
`176.108.138.1/32`, so TCP/22 timed out before authentication.
The intended durable path is `.github/workflows/gcp-iap-operator.yml`, using The intended secondary path is `.github/workflows/gcp-iap-operator.yml`, using
short-lived GitHub OIDC, IAP, OS Login, and fixed reviewed operations. It is short-lived GitHub OIDC, IAP, OS Login, and fixed reviewed operations. It is
merged but not bootstrapped: workflow run `29208215340` failed at auth with merged but not bootstrapped: workflow run `29208215340` failed at auth with
`invalid_target` because provider `teleo-iap-operator` is absent, disabled, or `invalid_target` because provider `teleo-iap-operator` is absent, disabled, or
@ -55,39 +54,34 @@ variable, then unset it.
Do not describe a passing Hermes memory sync as canonical KB parity. Do not describe a passing Hermes memory sync as canonical KB parity.
## Current Verified State - 2026-07-12 ## Current Verified State - 2026-07-14
- GCP `teleo_canonical` matches the captured VPS canonical database across - The newest captured VPS database has `39` tables and `52,167` rows, including
`39/39` tables and `52,164/52,164` rows. claims `1837`, sources `4145`, claim evidence `4670`, claim edges `4916`, and
- Exact high-signal counts are claims `1837`, sources `4145`, claim evidence proposals `29`.
`4670`, claim edges `4916`, and proposals `26`. - A disposable private-TLS GCP clone restored that snapshot with exact
- Rowsets, schemas, columns, constraints, indexes, functions, types, triggers, `39/39`-table and `52,167/52,167`-row parity. Rowsets, schema objects, roles,
views, policies, required roles, extensions, and performance checks have zero extensions, constraints, and performance checks had zero mismatches.
mismatches. `pgcrypto` and `kb_apply` are present. - A real no-send GCP Hermes turn received an ID-free claim challenge, performed
- The GCP gateway remained active/running at PID `148735`, `NRestarts=0`, with `search`, `show`, `evidence`, and `edges`, retrieved the expected claim and
start time `2026-07-09 07:00:04 UTC` throughout restore, swap, and read tests. both source rows, and passed `18/18` runtime checks plus `6/6` reasoning
- The reviewed GCP `teleo-kb` wrapper, Cloud SQL tool, and bridge skill were outcomes. It did not send Telegram or write the DB.
synchronized without a service restart. All six `DC-01` through `DC-06` - `status` and zero-hit search work on a canonical-only clone without the
database-read routes pass. optional `teleo_restore` audit schema. All read commands emit deterministic
- Generated candidate databases and temporary client/operation directories were retrieval receipts.
removed. `teleo_canonical_pre_20260712t1905z` is retained disabled, with zero - The generated clone was deleted. The retained rollback database remains
connections, as the verified pre-swap rollback point. connection-disabled with zero sessions. The GCP gateway remained PID
- A real no-send model replay returned six replies and nominally scored `6/6`, `148735`, `NRestarts=0`, active/running throughout.
but it is rejected as m3taversal-standard proof because `DC-03` and `DC-05` - Persistent GCP `teleo_canonical` remains the older staging copy measured at
printed zero canonical counts. The real database has `52,164` rows and `26` proposals. It has not been promoted or cut over.
`1837/4145/4916/4670/26`.
- The harness now enables clone-bound, default-read-only `teleo-kb status`, runs
under the Hermes virtualenv, and rejects any printed count that differs from
the canonical status receipt. The hardened rerun and clone cleanup wait on
the exact operator-access gate below.
Primary retained proof: Primary retained proof:
- `docs/reports/leo-working-state-20260709/gcp-canonical-parity-live-20260712.json` - `docs/reports/leo-working-state-20260709/gcp-db-first-working-leo-20260714.md`
- `docs/reports/leo-working-state-20260709/gcp-cory-model-replay-current.json` - `docs/reports/leo-working-state-20260709/gcp-db-first-restore-current.json`
- `docs/reports/leo-working-state-20260709/gcp-operator-access-blocker-current.json` - `docs/reports/leo-working-state-20260709/gcp-db-first-parity-current.json`
- root task output `outputs/gcp-staging-canonical-parity-20260712T1905Z/` - `docs/reports/leo-working-state-20260709/gcp-db-first-blind-claim-current.json`
- root task output `outputs/gcp-cory-model-replay-20260712T1940Z/` - `docs/reports/leo-working-state-20260709/gcp-db-first-cleanup-current.json`
## Required Parity Rows ## Required Parity Rows
@ -108,10 +102,16 @@ Use:
- `ops/capture_vps_canonical_postgres_snapshot.py` for a single source dump and - `ops/capture_vps_canonical_postgres_snapshot.py` for a single source dump and
manifest; manifest;
- `ops/restore_gcp_generated_postgres_snapshot.py restore --execute` for a
receipt-bound private-TLS restore into a bounded `teleo_clone_*` database;
- `ops/restore_gcp_generated_postgres_snapshot.py cleanup --execute` for exact
clone cleanup with live-service and rollback readback;
- `ops/postgres_parity_manifest.sql` for row/catalog/role/performance readback; - `ops/postgres_parity_manifest.sql` for row/catalog/role/performance readback;
- `ops/verify_postgres_parity_manifest.py --scope gcp_staging` for exact parity; - `ops/verify_postgres_parity_manifest.py --scope gcp_staging` for exact parity;
- `scripts/run_gcp_generated_db_direct_claim_suite.py` for the adapter-free, - `scripts/run_gcp_generated_db_direct_claim_suite.py` for the adapter-free,
read-only, no-send six-response replay against a generated `teleo_clone_*`. read-only, no-send six-response replay against a generated `teleo_clone_*`.
- `scripts/run_gcp_generated_db_blind_claim_canary.py` for the bounded ID-free
claim challenge, source receipt, reasoning, no-write, and cleanup proof.
The replay must use the Hermes virtualenv, not system Python: The replay must use the Hermes virtualenv, not system Python:
@ -136,20 +136,14 @@ The strongest accepted claim requires exact DB parity plus real no-send model
replies with truthful counts and cleanup. It still does not prove Telegram replies with truthful counts and cleanup. It still does not prove Telegram
delivery, GCP canonical mutation, ongoing replication, or production cutover. delivery, GCP canonical mutation, ongoing replication, or production cutover.
## Exact Current Gate ## Current Access And Next Action
All autonomous alternatives have been exhausted: privileged local `gcloud` Direct SSH and passwordless sudo are currently working. Local `gcloud` still
requires Google password reauthentication; direct SSH misses the firewall requires account reauthentication for control-plane metadata, so the current
`/32`; the dedicated WIF provider is not bootstrapped; the existing private-IP/TLS proof comes from a live database connection while the public-IP-
`living-ip-github` provider can authenticate only an artifact service account disabled control-plane receipt remains dated 2026-07-12.
that lacks Compute/IAM/Secret Manager/Cloud SQL permissions; no Mullvad or
Tailscale route exposes the allowed office ISP address.
CTA: in Chrome, open the existing tab titled `Google Cloud Platform` for The next production decision is not another restore drill. It is whether to
`billy@livingip.xyz`, enter the current Google password, click `Next`, then tell promote a newly verified snapshot into persistent GCP `teleo_canonical` and
Codex exactly `GCP reauthenticated`. Do not send the password. repoint the production read adapter. Until that decision is explicit, keep GCP
classified as staging and do not expose Cloud SQL publicly.
After that, rotate only the SSH `/32`, bootstrap and live-test the IAP operator,
run the hardened replay, delete `teleo_clone_cory_20260712t1940z`, remove its
run directory, and verify the disabled rollback database still has zero
connections.

View file

@ -24,11 +24,11 @@ Orient the worker before action. Build a current, proof-linked understanding of
- Leo is the operator-facing agent expected to answer in Telegram, remember operator context, reason from canonical KB state, stage concrete KB changes, and support approved changes becoming canonical DB rows with proof. - Leo is the operator-facing agent expected to answer in Telegram, remember operator context, reason from canonical KB state, stage concrete KB changes, and support approved changes becoming canonical DB rows with proof.
- The immediate July 9 issue is not generic bot liveness. It is m3taversal's expectation that approved KB changes move beyond proposal state when appropriate. - The immediate July 9 issue is not generic bot liveness. It is m3taversal's expectation that approved KB changes move beyond proposal state when appropriate.
- The VPS is the currently proven Telegram-visible Leo surface. - The VPS is the currently proven Telegram-visible Leo surface.
- GCP is a separate lane. The live gateway, private Cloud SQL canonical rows, - GCP is a separate lane. Direct passwordless SSH and `sudo -n` work as of
and exact VPS-to-GCP database parity are proven. Direct passwordless SSH was 2026-07-14. The newest VPS DB was restored to a disposable private Cloud SQL
proven but is not currently on-demand because its firewall `/32` no longer clone with exact parity, and a real ID-free no-send reasoning turn passed.
matches this Mac's egress, and the dedicated OIDC/IAP path is not bootstrapped. Persistent GCP `teleo_canonical` is still an older staging copy; Telegram
Telegram delivery, GCP canonical mutation, ongoing replication, and cutover delivery, GCP canonical mutation, ongoing replication, promotion, and cutover
remain separate proof rows. remain separate proof rows.
## Architecture Boundaries ## Architecture Boundaries
@ -48,16 +48,16 @@ Orient the worker before action. Build a current, proof-linked understanding of
From the repo root, read: From the repo root, read:
1. `docs/reports/leo-working-state-20260709/working-leo-current-proof-20260712.md` 1. `docs/reports/leo-working-state-20260709/current-truth-index.md`
2. `docs/reports/leo-working-state-20260709/current-truth-index.md` 2. `docs/reports/leo-working-state-20260709/gcp-db-first-working-leo-20260714.md`
3. `docs/reports/leo-working-state-20260709/operator-surface-map.md` 3. `docs/reports/leo-working-state-20260709/gcp-db-first-parity-current.json`
4. `docs/reports/leo-working-state-20260709/working-leo-definition-20260709.md` 4. `docs/reports/leo-working-state-20260709/gcp-db-first-blind-claim-current.json`
5. `docs/reports/leo-working-state-20260709/cory-expected-working-leo-outcomes-20260709.md` 5. `docs/reports/leo-working-state-20260709/gcp-db-first-cleanup-current.json`
6. `docs/reports/leo-working-state-20260709/approve-claim-clone-canary-current.md` 6. `docs/reports/leo-working-state-20260709/operator-surface-map.md`
7. `docs/reports/leo-working-state-20260709/gcp-cloud-sql-t3-live-readonly-current.md` 7. `docs/reports/leo-working-state-20260709/working-leo-definition-20260709.md`
8. `docs/reports/leo-working-state-20260709/gcp-canonical-parity-live-20260712.json` 8. `.agents/skills/working-leo-m3taversal-outcomes/SKILL.md`
9. `docs/reports/leo-working-state-20260709/gcp-cory-model-replay-current.json` 9. `docs/reports/leo-working-state-20260709/approve-claim-clone-canary-current.md`
10. `docs/reports/leo-working-state-20260709/gcp-operator-access-blocker-current.json` 10. `docs/reports/leo-working-state-20260709/gcp-cloud-sql-t3-live-readonly-current.md`
## Required Status Split ## Required Status Split
@ -79,11 +79,11 @@ Do not collapse GCP demo readiness into Telegram completion. Do not collapse pro
- Do not production-apply DB packets unless explicitly authorized. - Do not production-apply DB packets unless explicitly authorized.
- Do not expose secret contents. - Do not expose secret contents.
- Do not treat an old summary as current if a fresh readback is cheap. - Do not treat an old summary as current if a fresh readback is cheap.
- Try `ssh teleo-gcp-staging` for passwordless GCP VM access, but verify the - Use `ssh -o BatchMode=yes teleo-gcp-staging true` to preflight passwordless
current public egress matches the firewall `/32`. If it does not, follow the GCP VM access. If the firewall route drifts, rediscover the authenticated
exact gate in `gcp-operator-access-blocker-current.json`; do not pretend the non-secret route before declaring a blocker. The OIDC/IAP workflow is still
merged OIDC/IAP workflow is bootstrapped. Do not print the Cloud SQL or Google not bootstrapped. Do not print the Cloud SQL or Google password or create
password or create IAM/users inside a read-only lane. IAM/users inside a read-only lane.
- A benchmark score is not sufficient when an answer contradicts the DB receipt. - A benchmark score is not sufficient when an answer contradicts the DB receipt.
Compare every printed count to `teleo-kb status` or the canonical manifest. Compare every printed count to `teleo-kb status` or the canonical manifest.

View file

@ -110,37 +110,38 @@ that would change the proof:
End no-context answers with exactly one `Next proof-changing follow-up:` line. End no-context answers with exactly one `Next proof-changing follow-up:` line.
## Current Verdict - 2026-07-13 ## Current Verdict - 2026-07-14
Use `not fully yet` for the whole m3taversal-standard question until every row below Use `not fully yet` for the whole m3taversal-standard question until every row below
is green at its required tier: is green at its required tier:
- VPS runtime and restart survival: proven. - VPS runtime, database-first direct questions, and intentional restart
- VPS no-send `DC-01` through `DC-06`: three post-deploy clean-session trials survival are proven at no-send/runtime tier. The newest captured VPS canonical
strict-score `6/6` (`18/18` replies) with unchanged DB/service state and DB has `39` tables, `52,167` rows, and `29` proposals.
complete temporary-profile cleanup. - One natural ID-free VPS claim challenge now passes the complete bounded chain:
- Telegram-visible open-ended read-only behavior and conversation memory: discover claim, inspect body/evidence/edges, challenge shallow support,
proven for the earlier `OE-01` through `OE-05` suite, but that narrower suite propose candidate claims, and preserve review before any write. This does not
does not establish current broad reliability. erase the older broad 12-prompt failure; broad repeated reliability remains a
- Telegram-visible no-context direct-claim suite: all six prompts/replies are separate benchmark row.
captured with unchanged DB/service state, but the hardened score is `5/6`. - A fresh VPS snapshot was restored into a disposable private GCP Cloud SQL
`DC-05` proposes an incompatible legacy `add_edge` apply target. database with exact `39/39`-table and `52,167/52,167`-row parity, zero catalog
- Canonical apply primitive and approved bundle lifecycle: proven in isolation; mismatches, and unchanged production service state.
broad production packet application is not authorized or executed. - The same GCP ID-free challenge passed `18/18` runtime checks and `6/6`
- Source composition: deterministic fixture proof plus a full-data, no-send, reasoning outcomes through the real Hermes runner. It used four successful
current-VPS clone checkpoint are proven. Arbitrary production document/tweet read-only, receipted calls and retrieved the expected claim plus both source
ingestion is not proven. rows. No Telegram send or database write occurred.
- Broad blind VPS behavior: operationally clean but semantically failed. - The GCP helper now gives deterministic receipts for every read, handles a
Twelve of twelve prompts returned, DB/service state stayed unchanged, and the canonical-only restored DB without the optional audit schema, and returns an
temporary profile was removed; independent strict judges accepted only honest zero-hit result instead of crashing.
`1/12` and `2/12` outright. Failures include invented current schema fields - The disposable GCP database and temporary profiles/processes were removed.
and edge types, handler proof described as Telegram-live, incorrect runtime Persistent GCP `teleo_canonical` remains the older staging copy and has not
memory boundaries, and temporary memory treated as source provenance. been promoted or cut over.
- GCP canonical DB parity and six DB-read routes: proven. A later adapter-free - Earlier Telegram-visible open-ended behavior and conversation memory are
model replay passed `6/6` with exact count consistency and unchanged state. retained proof, but there is no current Telegram-visible run of this full
Durable on-demand operator access and cleanup of the retained replay clone challenge-to-candidate-claims flow.
remain open because the expected operator service accounts are absent and the - Canonical apply primitives are proven in isolation. No proposal from this
privileged Google account requires human password/MFA reauthentication. claim challenge was staged, reviewed, or production-applied, and broad
arbitrary document/post reconstruction from original sources remains open.
Do not answer `yes` merely because all repo tests pass. The final user-facing Do not answer `yes` merely because all repo tests pass. The final user-facing
proof is a visible Telegram conversation plus truthful canonical row readback. proof is a visible Telegram conversation plus truthful canonical row readback.

View file

@ -49,11 +49,13 @@ jobs:
ops/check_gcp_service_communications.py \ ops/check_gcp_service_communications.py \
ops/plan_gcp_iam_split.py \ ops/plan_gcp_iam_split.py \
ops/redact_sqlite_postgres_restore_canary.py \ ops/redact_sqlite_postgres_restore_canary.py \
ops/restore_gcp_generated_postgres_snapshot.py \
ops/sqlite_to_postgres_dump.py \ ops/sqlite_to_postgres_dump.py \
ops/verify_gcp_cloudsql_restore_readback.py \ ops/verify_gcp_cloudsql_restore_readback.py \
ops/verify_postgres_parity_manifest.py \ ops/verify_postgres_parity_manifest.py \
telegram/approvals.py \ telegram/approvals.py \
hermes-agent/leoclean-bin/kb_tool.py \ hermes-agent/leoclean-bin/kb_tool.py \
hermes-agent/leoclean-bin/cloudsql_memory_tool.py \
scripts/check_crabbox_ci_contract.py \ scripts/check_crabbox_ci_contract.py \
scripts/check_llm_refinement_contract.py \ scripts/check_llm_refinement_contract.py \
scripts/build_working_leo_m3taversal_outcome_sandbox.py \ scripts/build_working_leo_m3taversal_outcome_sandbox.py \
@ -62,6 +64,7 @@ jobs:
scripts/replay_decision_engine_eval.py \ scripts/replay_decision_engine_eval.py \
scripts/prove_phase1b_local.py \ scripts/prove_phase1b_local.py \
scripts/run_gcp_generated_db_direct_claim_suite.py \ scripts/run_gcp_generated_db_direct_claim_suite.py \
scripts/run_gcp_generated_db_blind_claim_canary.py \
scripts/run_leo_direct_claim_handler_suite.py \ scripts/run_leo_direct_claim_handler_suite.py \
scripts/run_leo_m3taversal_oos_handler_suite.py \ scripts/run_leo_m3taversal_oos_handler_suite.py \
scripts/verify_leo_db_first_oos_canary.py \ scripts/verify_leo_db_first_oos_canary.py \
@ -85,6 +88,7 @@ jobs:
tests/test_gcp_iam_split_plan.py \ tests/test_gcp_iam_split_plan.py \
tests/test_gcp_readiness_workflow.py \ tests/test_gcp_readiness_workflow.py \
tests/test_gcp_generated_db_direct_claim_suite.py \ tests/test_gcp_generated_db_direct_claim_suite.py \
tests/test_gcp_generated_db_blind_claim_canary.py \
tests/test_hermes_leoclean_kb_bridge_source.py \ tests/test_hermes_leoclean_kb_bridge_source.py \
tests/test_hermes_leoclean_skill_surfaces.py \ tests/test_hermes_leoclean_skill_surfaces.py \
tests/test_leo_behavior_manifest.py \ tests/test_leo_behavior_manifest.py \
@ -95,6 +99,7 @@ jobs:
tests/test_working_leo_m3taversal_oos_benchmark.py \ tests/test_working_leo_m3taversal_oos_benchmark.py \
tests/test_working_leo_open_ended_benchmark.py \ tests/test_working_leo_open_ended_benchmark.py \
tests/test_phase1b_end_to_end.py \ tests/test_phase1b_end_to_end.py \
tests/test_restore_gcp_generated_postgres_snapshot.py \
tests/test_sqlite_to_postgres_dump.py \ tests/test_sqlite_to_postgres_dump.py \
tests/test_sqlite_postgres_restore_canary_capsule.py \ tests/test_sqlite_postgres_restore_canary_capsule.py \
tests/test_eval_parse.py \ tests/test_eval_parse.py \

View file

@ -2,6 +2,33 @@
Use this file before making status claims. Prefer fresh VPS/GCP readbacks when cheap; this directory is the retained July 9 evidence snapshot committed to the Teleo infrastructure repo. Use this file before making status claims. Prefer fresh VPS/GCP readbacks when cheap; this directory is the retained July 9 evidence snapshot committed to the Teleo infrastructure repo.
## GCP Database-First Addendum - 2026-07-14 10:07 UTC
This addendum supersedes the GCP-pending statements in the earlier July 14
section below.
- The current VPS snapshot (`39` tables, `52,167` rows, `29` proposals) was
restored into a disposable private-TLS GCP Cloud SQL database with exact
table/row/catalog/role/extension/performance parity.
- A real no-send GCP Hermes turn answered an ID-free challenge to a shallow
claim. It ran `search`, `show`, `evidence`, and `edges`, retrieved the exact
claim plus both expected source rows, passed `18/18` runtime checks and `6/6`
reasoning outcomes, made no DB write, and sent no Telegram message.
- The disposable clone was deleted. The rollback DB remains disabled with zero
connections. The live GCP gateway stayed PID `148735`, `NRestarts=0`.
- Persistent GCP `teleo_canonical` is still the older staging copy measured at
`52,164` rows and `26` proposals. No promotion or production cutover occurred.
- Fast snapshot restore is now proven. Full reconstruction from every original
document/post/repository artifact remains incomplete.
Newest GCP database-first artifacts:
- `gcp-db-first-working-leo-20260714.md`
- `gcp-db-first-restore-current.json`
- `gcp-db-first-parity-current.json`
- `gcp-db-first-blind-claim-current.json`
- `gcp-db-first-cleanup-current.json`
## Live Addendum - 2026-07-14 ## Live Addendum - 2026-07-14
The current canonical handoff is The current canonical handoff is

File diff suppressed because one or more lines are too long

View file

@ -0,0 +1,40 @@
{
"artifact": "gcp_generated_postgres_snapshot_cleanup",
"database": {
"clone_database_remaining": 0,
"existed_before": false
},
"generated_at_utc": "2026-07-14T10:07:17.346308+00:00",
"live_service": {
"after": {
"ActiveState": "active",
"ExecMainStartTimestamp": "Thu 2026-07-09 07:00:04 UTC",
"MainPID": "148735",
"NRestarts": "0",
"SubState": "running"
},
"before": {
"ActiveState": "active",
"ExecMainStartTimestamp": "Thu 2026-07-09 07:00:04 UTC",
"MainPID": "148735",
"NRestarts": "0",
"SubState": "running"
},
"unchanged": true
},
"rollback_database": {
"connections": 0,
"datallowconn": false,
"exists": true
},
"run_id": "gcp-restore-dbfirst-20260714t090758z",
"safety": {
"live_database_named": false,
"live_profile_modified": false,
"live_service_restarted": false,
"secret_persisted": false,
"telegram_message_sent": false
},
"status": "pass",
"target_database": "teleo_clone_dbfirst_20260714t090758z"
}

View file

@ -0,0 +1,133 @@
{
"artifact": "canonical_postgres_parity_verification",
"details": {
"application_role_mismatches": {},
"performance": [
{
"query": "claim_edges_by_claim_id",
"source_ms": 0.043,
"source_ratio": 0.49000000000000005,
"target_ms": 2.45
},
{
"query": "claim_evidence_by_claim_id",
"source_ms": 0.062,
"source_ratio": 0.1742,
"target_ms": 0.871
},
{
"query": "count_claims",
"source_ms": 0.23,
"source_ratio": 0.09140000000000001,
"target_ms": 0.457
},
{
"query": "pending_proposals_latest",
"source_ms": 0.015,
"source_ratio": 0.0018,
"target_ms": 0.009
}
],
"performance_problems": [],
"required_extension_mismatches": {},
"source_connection": {
"database_collation": "en_US.utf8",
"database_ctype": "en_US.utf8",
"server_address": null,
"server_port": null,
"ssl": false,
"ssl_version": null
},
"source_database": "teleo",
"source_table_count": 39,
"source_total_rows": 52167,
"structural_hashes": {
"columns": {
"source": "50b502967aa85b88bd204c7b9035de60bf21a6a6973817e8b8121d0cd9faab81",
"target": "50b502967aa85b88bd204c7b9035de60bf21a6a6973817e8b8121d0cd9faab81"
},
"constraints": {
"source": "8b26d26fda56f09325e0603aae48051da08dfc6c35322fa88880acdabb8ad859",
"target": "8b26d26fda56f09325e0603aae48051da08dfc6c35322fa88880acdabb8ad859"
},
"functions": {
"source": "4bddee626b3ab62af236493001d2a09f2d56c084beb7e6c69cd082ce4aba155b",
"target": "4bddee626b3ab62af236493001d2a09f2d56c084beb7e6c69cd082ce4aba155b"
},
"indexes": {
"source": "5c4f7567e9ad0f25da0ab15f650a1005397ae9c727af1ab41e2675c46a5ebce5",
"target": "5c4f7567e9ad0f25da0ab15f650a1005397ae9c727af1ab41e2675c46a5ebce5"
},
"policies": {
"source": "4f53cda18c2baa0c0354bb5f9a3ecbe5ed12ab4d8e11ba873c2f11161202b945",
"target": "4f53cda18c2baa0c0354bb5f9a3ecbe5ed12ab4d8e11ba873c2f11161202b945"
},
"schemas": {
"source": "2fdf551b403db2d79813c8fd2869dadfeba261fac2aaa718b355cd7414640979",
"target": "2fdf551b403db2d79813c8fd2869dadfeba261fac2aaa718b355cd7414640979"
},
"sequences": {
"source": "4f53cda18c2baa0c0354bb5f9a3ecbe5ed12ab4d8e11ba873c2f11161202b945",
"target": "4f53cda18c2baa0c0354bb5f9a3ecbe5ed12ab4d8e11ba873c2f11161202b945"
},
"triggers": {
"source": "4d5f3394480c8410035dac5457e144c7db80fa0820ff49beb93aa819a7282c64",
"target": "4d5f3394480c8410035dac5457e144c7db80fa0820ff49beb93aa819a7282c64"
},
"types": {
"source": "365e82ac205319d25d14c8b16026978f0e6023ae0c58767f700d7f2ea659b70b",
"target": "365e82ac205319d25d14c8b16026978f0e6023ae0c58767f700d7f2ea659b70b"
},
"views": {
"source": "5c33fa260f25dccffc214f795c66e3dbae89d91d0dbf8a98122ace1dcda46799",
"target": "5c33fa260f25dccffc214f795c66e3dbae89d91d0dbf8a98122ace1dcda46799"
}
},
"table_mismatches": [],
"target_connection": {
"database_collation": "en_US.utf8",
"database_ctype": "en_US.utf8",
"server_address": "10.61.0.3",
"server_port": 5432,
"ssl": true,
"ssl_version": "TLSv1.3"
},
"target_database": "teleo_clone_dbfirst_20260714t090758z",
"target_extra_application_roles": [],
"target_table_count": 39,
"target_total_rows": 52167
},
"error": null,
"generated_at_utc": "2026-07-14T09:14:02.603285+00:00",
"not_proven_by_this_artifact": [
"GCP staging Leo composition replay",
"ongoing replication after the manifest timestamps",
"production cutover"
],
"private_connectivity": {
"path": "/private/tmp/teleo-gcp-dbfirst-proof-20260714/connectivity.json",
"proof": {
"captured_at_utc": "2026-07-14T09:09:12.241790+00:00",
"evidence": {
"current_private_tls_manifest": "/private/tmp/teleo-gcp-dbfirst-proof-20260714/target-manifest.jsonl",
"current_private_tls_manifest_sha256": "6f834b1486af1e24193a1ded517b4cc9e64e19c68097c028e192dd298986855c",
"public_ip_disabled_control_plane_captured_at_utc": "2026-07-12T19:36:50.869980+00:00",
"public_ip_disabled_control_plane_receipt": "/Users/user/Documents/Codex/2026-07-09/019f34eb-d297-72d0-b7e2-b222d5515ab9-load/outputs/gcp-staging-canonical-parity-20260712T1905Z/final-connectivity-post-cleanup.json"
},
"private_connectivity": true,
"public_ip_disabled": true,
"server_address": "10.61.0.3",
"source_compute": "teleo-prod-1",
"ssl": true,
"ssl_version": "TLSv1.3",
"status": "pass",
"target_database": "teleo_clone_dbfirst_20260714t090758z"
},
"required": true
},
"problems": [],
"scope": "gcp_staging",
"source_manifest": "/private/tmp/teleo-gcp-dbfirst-source-20260714/source-manifest.jsonl",
"status": "pass",
"target_manifest": "/private/tmp/teleo-gcp-dbfirst-proof-20260714/target-manifest.jsonl"
}

View file

@ -0,0 +1,182 @@
{
"artifact": "gcp_generated_postgres_snapshot_restore",
"cleanup": {
"attempted": false,
"clone_database_remaining": 1,
"exact_command": "sudo python3 ops/restore_gcp_generated_postgres_snapshot.py cleanup --execute --target-db teleo_clone_dbfirst_20260714t090758z --run-id gcp-restore-dbfirst-20260714t090758z",
"required": true
},
"duration_seconds": 103.83768,
"error": null,
"generated_at_utc": "2026-07-14T09:09:12.241790+00:00",
"live_service": {
"after": {
"ActiveState": "active",
"ExecMainStartTimestamp": "Thu 2026-07-09 07:00:04 UTC",
"MainPID": "148735",
"NRestarts": "0",
"SubState": "running"
},
"before": {
"ActiveState": "active",
"ExecMainStartTimestamp": "Thu 2026-07-09 07:00:04 UTC",
"MainPID": "148735",
"NRestarts": "0",
"SubState": "running"
},
"unchanged": true
},
"parity": {
"details": {
"application_role_mismatches": {},
"performance": [
{
"query": "claim_edges_by_claim_id",
"source_ms": 0.043,
"source_ratio": 0.49000000000000005,
"target_ms": 2.45
},
{
"query": "claim_evidence_by_claim_id",
"source_ms": 0.062,
"source_ratio": 0.1742,
"target_ms": 0.871
},
{
"query": "count_claims",
"source_ms": 0.23,
"source_ratio": 0.09140000000000001,
"target_ms": 0.457
},
{
"query": "pending_proposals_latest",
"source_ms": 0.015,
"source_ratio": 0.0018,
"target_ms": 0.009
}
],
"performance_problems": [],
"required_extension_mismatches": {},
"source_connection": {
"database_collation": "en_US.utf8",
"database_ctype": "en_US.utf8",
"server_address": null,
"server_port": null,
"ssl": false,
"ssl_version": null
},
"source_database": "teleo",
"source_table_count": 39,
"source_total_rows": 52167,
"structural_hashes": {
"columns": {
"source": "50b502967aa85b88bd204c7b9035de60bf21a6a6973817e8b8121d0cd9faab81",
"target": "50b502967aa85b88bd204c7b9035de60bf21a6a6973817e8b8121d0cd9faab81"
},
"constraints": {
"source": "8b26d26fda56f09325e0603aae48051da08dfc6c35322fa88880acdabb8ad859",
"target": "8b26d26fda56f09325e0603aae48051da08dfc6c35322fa88880acdabb8ad859"
},
"functions": {
"source": "4bddee626b3ab62af236493001d2a09f2d56c084beb7e6c69cd082ce4aba155b",
"target": "4bddee626b3ab62af236493001d2a09f2d56c084beb7e6c69cd082ce4aba155b"
},
"indexes": {
"source": "5c4f7567e9ad0f25da0ab15f650a1005397ae9c727af1ab41e2675c46a5ebce5",
"target": "5c4f7567e9ad0f25da0ab15f650a1005397ae9c727af1ab41e2675c46a5ebce5"
},
"policies": {
"source": "4f53cda18c2baa0c0354bb5f9a3ecbe5ed12ab4d8e11ba873c2f11161202b945",
"target": "4f53cda18c2baa0c0354bb5f9a3ecbe5ed12ab4d8e11ba873c2f11161202b945"
},
"schemas": {
"source": "2fdf551b403db2d79813c8fd2869dadfeba261fac2aaa718b355cd7414640979",
"target": "2fdf551b403db2d79813c8fd2869dadfeba261fac2aaa718b355cd7414640979"
},
"sequences": {
"source": "4f53cda18c2baa0c0354bb5f9a3ecbe5ed12ab4d8e11ba873c2f11161202b945",
"target": "4f53cda18c2baa0c0354bb5f9a3ecbe5ed12ab4d8e11ba873c2f11161202b945"
},
"triggers": {
"source": "4d5f3394480c8410035dac5457e144c7db80fa0820ff49beb93aa819a7282c64",
"target": "4d5f3394480c8410035dac5457e144c7db80fa0820ff49beb93aa819a7282c64"
},
"types": {
"source": "365e82ac205319d25d14c8b16026978f0e6023ae0c58767f700d7f2ea659b70b",
"target": "365e82ac205319d25d14c8b16026978f0e6023ae0c58767f700d7f2ea659b70b"
},
"views": {
"source": "5c33fa260f25dccffc214f795c66e3dbae89d91d0dbf8a98122ace1dcda46799",
"target": "5c33fa260f25dccffc214f795c66e3dbae89d91d0dbf8a98122ace1dcda46799"
}
},
"table_mismatches": [],
"target_connection": {
"database_collation": "en_US.utf8",
"database_ctype": "en_US.utf8",
"server_address": "10.61.0.3",
"server_port": 5432,
"ssl": true,
"ssl_version": "TLSv1.3"
},
"target_database": "teleo_clone_dbfirst_20260714t090758z",
"target_extra_application_roles": [],
"target_table_count": 39,
"target_total_rows": 52167
},
"problems": [],
"status": "pass",
"verifier": "ops.verify_postgres_parity_manifest.compare_manifests"
},
"phase": "retained_for_no_send_replay",
"rollback_database": {
"connections": 0,
"datallowconn": false,
"exists": true
},
"run_id": "gcp-restore-dbfirst-20260714t090758z",
"safety": {
"live_database_named": false,
"live_profile_modified": false,
"live_service_restarted": false,
"secret_persisted": false,
"telegram_message_sent": false
},
"source": {
"dump": {
"bytes": 13468277,
"path": "/home/billy_livingip_xyz/.teleo-gcp-restore-upload-20260714t085320z/source/teleo-canonical.dump",
"sha256": "92cc2cafd7575cd38f59745674dd3d18a1f2120d7f87821459b8b38544263f81"
},
"manifest": {
"bytes": 164734,
"path": "/home/billy_livingip_xyz/.teleo-gcp-restore-upload-20260714t085320z/source/source-manifest.jsonl",
"sha256": "efef2c1d2609879872e70e06758e6d59ce657686eb1e740560e4246fc11ec7c4"
}
},
"status": "pass",
"target": {
"connectivity": {
"private_connectivity": true,
"server_address": "10.61.0.3",
"server_port": 5432,
"source_compute": "teleo-prod-1",
"ssl": true,
"ssl_version": "TLSv1.3",
"target_database": "teleo_clone_dbfirst_20260714t090758z",
"transaction_read_only": "on"
},
"manifest_bytes": 164776,
"manifest_path": "/var/lib/teleo-gcp-restore-runs/gcp-restore-dbfirst-20260714t090758z/target-manifest.jsonl",
"manifest_sha256": "6f834b1486af1e24193a1ded517b4cc9e64e19c68097c028e192dd298986855c",
"restore_seconds": 61.761689
},
"target_database": "teleo_clone_dbfirst_20260714t090758z",
"toolchain": {
"compatible": true,
"pg_restore_bin": "/usr/lib/postgresql/16/bin/pg_restore",
"pg_restore_major": 16,
"pg_restore_version": "pg_restore (PostgreSQL) 16.14 (Debian 16.14-1.pgdg12+1)",
"source_postgres_major": 16
}
}

View file

@ -0,0 +1,148 @@
# GCP Database-First Working Leo Proof - 2026-07-14
## One-Line Result
Leo can now take an ID-free challenge to a real claim, find it in a freshly
rebuilt private GCP database, inspect its body/evidence/edges, explain why it is
shallow, propose better claims, and keep the user in the review loop before any
database change. The bounded no-send run passed `18/18` runtime checks and
`6/6` reasoning outcomes.
This is strong GCP no-send handler proof. It is not Telegram delivery, a
production knowledge apply, continuous replication, or a GCP production
cutover.
## Problem We Were Solving
The system mixed several meanings of "Leo learned": conversation history,
rendered identity files, deployed instructions, proposal rows, and canonical
knowledge. That made it hard to tell whether an answer came from current
database facts, stale runtime context, or benchmark-specific training.
The target is database-first behavior:
1. A discussion is candidate input, not hidden training.
2. Leo retrieves current canonical claims and evidence before asserting KB fact.
3. Leo can challenge weak knowledge and draft candidate improvements.
4. Candidate changes remain reviewable and replayable before canonical apply.
5. Every read and write has a receipt tied to exact database state.
6. The database can be copied and verified without retraining Leo.
## What Makes Leo Leo
| Layer | Change rate | Role | Canonical knowledge? |
| --- | --- | --- | --- |
| Base model and Hermes runtime | Slow/versioned | General language and tool-using ability | No |
| Deployed skills, tool wrappers, and routing config | Versioned deployment | Tell Leo when and how to query, reason, stage, and apply | No |
| Rendered `SOUL.md` and identity/context files | Periodically generated or deployed | Compact runtime context and identity | No; derived/runtime input |
| `public.*` in canonical PostgreSQL | Dynamic and durable | Live claims, sources, evidence, edges, personas, beliefs, strategies | Yes |
| `kb_stage.*` | Dynamic and review-gated | Candidate proposals, review state, and apply state | No until applied |
| Hermes `state.db` and session JSONL | Dynamic runtime continuity | Remembers conversations and operational state | No |
| Current message/session context | Ephemeral | Immediate user intent and iteration | No |
Chat does not update model weights. A conversation can affect the current
reply and session continuity, but durable collective learning should become a
source-bound proposal and, after review and apply, canonical PostgreSQL rows.
## What Changed
### 1. Current VPS knowledge can be rebuilt quickly in GCP
`ops/restore_gcp_generated_postgres_snapshot.py` now restores a reviewed
PostgreSQL 16 custom dump into a bounded `teleo_clone_*` database over private
TLS. It verifies the source hash, client version, schema, rows, constraints,
roles, extensions, performance, rollback state, and live-service invariants.
Failure paths automatically remove partial clones.
This is a fast exact restore from the current canonical database. It is not yet
a full source-of-origin rebuild from every document, URL, post, and repository
artifact.
### 2. GCP retrieval is deterministic and source-bound
`cloudsql_memory_tool.py` now returns a retrieval receipt containing semantic
and artifact-state hashes, database identity, WAL consistency, claim IDs, and
source IDs. Receipts cover all read commands, including proposal and
decision-matrix reads.
Clean canonical clones no longer require the optional legacy `teleo_restore`
audit schema. `status` works without it, and a zero-hit search returns an honest
empty canonical result instead of crashing.
### 3. The real reasoning path was proved, not inferred
The exact prompt supplied no row IDs:
> Our claim that AI sandbagging creates M&A liability feels shallow. Without me
> giving you a claim ID, inspect the live claim and what actually supports it.
> Tell me what is weak, what new claim or claims you would propose, and how you
> would iterate with me before anything becomes live. Do not change the database.
The GCP Hermes turn performed exactly four successful read-only calls:
`search`, `show`, `evidence`, and `edges`. It retrieved claim
`2a7ae257-d01d-46f4-b813-63f81bb9c7c7` and both expected source rows, explained
why the legal claim was over-bundled and weakly grounded, proposed narrower
candidate claims, and asked for review before staging or apply.
The reply used the exact `m3taversal` handle contract and no forbidden alias.
No Telegram message was sent and no DB write was attempted.
### 4. The experiment cleaned itself up
The generated clone was deleted. A second cleanup receipt reports zero
remaining target databases. The disabled rollback database still has zero
connections. The live GCP Leo service remained PID `148735`, active/running,
with `NRestarts=0` throughout restore, model replay, and cleanup.
## VPS Versus GCP
| Surface | Current result | Meaning |
| --- | --- | --- |
| VPS production DB | `39` tables, `52,167` rows, `29` proposals at capture | Newest measured canonical source for this run |
| GCP persistent `teleo_canonical` | Previously verified at `39` tables, `52,164` rows, `26` proposals | Older staging copy; not promoted or cut over |
| Disposable GCP clone | Exact `39/39`, `52,167/52,167`, zero mismatches | Proves current VPS state can be recreated privately in GCP |
| GCP Hermes no-send turn | `18/18` checks, `6/6` outcomes | Proves the rebuilt DB works through the real reasoning/tool path |
| Disposable clone after test | Absent | Proves lifecycle cleanup; it is not a fourth persistent source of truth |
This is not three Git branches. The persistent divergence is operational: the
VPS canonical DB is newer than the still-staging GCP canonical DB, while the
GitHub repository owns the code and deployment instructions. Reconciliation
means choosing an authoritative DB snapshot, restoring/verifying GCP from it,
then explicitly promoting or cutting over. It does not mean merging database
rows as if they were source-code branches.
## Proof Receipts
| Receipt | SHA-256 | What it proves |
| --- | --- | --- |
| `gcp-db-first-restore-current.json` | `06591cb97c7108f7932d042912e2dcad9eb54bbca5cde72ad53642beb79aa89c` | Private restore completed and live service stayed unchanged |
| `gcp-db-first-parity-current.json` | `0173ee6707016e8412e6dd4326d61f71b6ef862bbc5b819079d62680676729f2` | Schema, row, role, extension, and performance parity passed |
| `gcp-db-first-blind-claim-current.json` | `8a7cc3c1814eb385e858f12ef7579cd4524f37886c03fc6d9c61624b6aff2a52` | Real GCP Hermes reasoning passed with source-bound read receipts |
| `gcp-db-first-cleanup-current.json` | `cac7e34f45653fabb696d911b6ccc921d3f291bf8cbe05a429d757e717dcafb4` | Clone absent, rollback disabled, service unchanged |
## What This Unlocks
- We can test database changes by rebuilding a disposable copy instead of
repeatedly changing Leo's prompt or session memory.
- We can compare answers against exact claim/evidence/source rows and detect
when a response came from unsupported runtime context.
- We can move the same database-first behavior from VPS to GCP without exposing
Cloud SQL publicly.
- We have a bounded path for future out-of-sample benchmarks: vague question,
fresh retrieval, claim challenge, candidate proposal, user iteration, guarded
apply, and row-level postflight.
## Still Not Proven
1. The same full challenge is visible in a current Telegram conversation.
2. A real user-approved candidate from this reasoning loop is staged, reviewed,
applied, and read back end to end on production.
3. The entire canonical DB can be reconstructed from original source documents
and repository artifacts rather than restored from a database snapshot.
4. VPS changes continuously replicate to GCP or GCP has been formally promoted.
5. Broad reliability across many unseen claim domains and repeated trials.
The next product-level proof is one natural Telegram challenge that reaches
the same database-grounded reasoning, followed by one explicitly approved
proposal lifecycle with exact source/claim/evidence/edge receipts. Those are
separate authorization and production-mutation steps; neither occurred here.

View file

@ -10,6 +10,7 @@ shadow schema in Cloud SQL. Proposal commands write staged review records into
from __future__ import annotations from __future__ import annotations
import argparse import argparse
import hashlib
import json import json
import os import os
import re import re
@ -17,14 +18,76 @@ import subprocess
from typing import Any from typing import Any
STOPWORDS = { STOPWORDS = {
"a", "about", "after", "all", "an", "and", "are", "as", "at", "be", "before", "a",
"by", "can", "cloud", "context", "do", "does", "for", "from", "gcp", "give", "about",
"how", "i", "in", "into", "is", "it", "leo", "me", "memory", "of", "on", "or", "after",
"our", "production", "should", "source", "that", "the", "their", "this", "to", "all",
"using", "what", "when", "where", "which", "who", "why", "with", "you", "your", "an",
"and",
"are",
"as",
"at",
"be",
"before",
"by",
"can",
"cloud",
"context",
"do",
"does",
"for",
"from",
"gcp",
"give",
"how",
"i",
"in",
"into",
"is",
"it",
"leo",
"me",
"memory",
"of",
"on",
"or",
"our",
"production",
"should",
"source",
"that",
"the",
"their",
"this",
"to",
"using",
"what",
"when",
"where",
"which",
"who",
"why",
"with",
"you",
"your",
} }
DEFAULT_CLAIM_BASE_URL = "https://leo.livingip.xyz" DEFAULT_CLAIM_BASE_URL = "https://leo.livingip.xyz"
RETRIEVAL_RECEIPT_SCHEMA = "livingip.teleoKbRetrievalReceipt.v1"
RECEIPTED_READ_COMMANDS = frozenset(
{
"search",
"context",
"show",
"evidence",
"edges",
"status",
"list-proposals",
"search-proposals",
"show-proposal",
"decision-matrix-status",
}
)
def parse_args() -> argparse.Namespace: def parse_args() -> argparse.Namespace:
@ -54,38 +117,52 @@ def parse_args() -> argparse.Namespace:
search = sub.add_parser("search", help="Search restored Cloud SQL memory rows.") search = sub.add_parser("search", help="Search restored Cloud SQL memory rows.")
search.add_argument("query") search.add_argument("query")
search.add_argument("--limit", type=int, default=8) search.add_argument("--limit", type=int, default=8)
search.add_argument("--context-limit", type=int, default=argparse.SUPPRESS, help="Accepted for teleo-kb compatibility; ignored.") search.add_argument(
"--context-limit", type=int, default=argparse.SUPPRESS, help="Accepted for teleo-kb compatibility; ignored."
)
search.add_argument("--format", choices=["markdown", "json"], default=argparse.SUPPRESS) search.add_argument("--format", choices=["markdown", "json"], default=argparse.SUPPRESS)
search.add_argument("--redacted", action="store_true", default=argparse.SUPPRESS) search.add_argument("--redacted", action="store_true", default=argparse.SUPPRESS)
search.add_argument("--include-excerpts", "--raw-ok", dest="include_excerpts", action="store_true", default=argparse.SUPPRESS) search.add_argument(
"--include-excerpts", "--raw-ok", dest="include_excerpts", action="store_true", default=argparse.SUPPRESS
)
context = sub.add_parser("context", help="Build an agent-ready memory context bundle.") context = sub.add_parser("context", help="Build an agent-ready memory context bundle.")
context.add_argument("query") context.add_argument("query")
context.add_argument("--limit", type=int, default=8) context.add_argument("--limit", type=int, default=8)
context.add_argument("--context-limit", type=int, default=argparse.SUPPRESS, help="Accepted for teleo-kb compatibility; ignored.") context.add_argument(
"--context-limit", type=int, default=argparse.SUPPRESS, help="Accepted for teleo-kb compatibility; ignored."
)
context.add_argument("--format", choices=["markdown", "json"], default=argparse.SUPPRESS) context.add_argument("--format", choices=["markdown", "json"], default=argparse.SUPPRESS)
context.add_argument("--redacted", action="store_true", default=argparse.SUPPRESS) context.add_argument("--redacted", action="store_true", default=argparse.SUPPRESS)
context.add_argument("--include-excerpts", "--raw-ok", dest="include_excerpts", action="store_true", default=argparse.SUPPRESS) context.add_argument(
"--include-excerpts", "--raw-ok", dest="include_excerpts", action="store_true", default=argparse.SUPPRESS
)
show = sub.add_parser("show", help="Show one canonical claim with evidence and edges.") show = sub.add_parser("show", help="Show one canonical claim with evidence and edges.")
show.add_argument("claim_id") show.add_argument("claim_id")
show.add_argument("--format", choices=["markdown", "json"], default=argparse.SUPPRESS) show.add_argument("--format", choices=["markdown", "json"], default=argparse.SUPPRESS)
show.add_argument("--redacted", action="store_true", default=argparse.SUPPRESS) show.add_argument("--redacted", action="store_true", default=argparse.SUPPRESS)
show.add_argument("--include-excerpts", "--raw-ok", dest="include_excerpts", action="store_true", default=argparse.SUPPRESS) show.add_argument(
"--include-excerpts", "--raw-ok", dest="include_excerpts", action="store_true", default=argparse.SUPPRESS
)
evidence = sub.add_parser("evidence", help="Show canonical evidence rows for one claim.") evidence = sub.add_parser("evidence", help="Show canonical evidence rows for one claim.")
evidence.add_argument("claim_id") evidence.add_argument("claim_id")
evidence.add_argument("--limit", type=int, default=8) evidence.add_argument("--limit", type=int, default=8)
evidence.add_argument("--format", choices=["markdown", "json"], default=argparse.SUPPRESS) evidence.add_argument("--format", choices=["markdown", "json"], default=argparse.SUPPRESS)
evidence.add_argument("--redacted", action="store_true", default=argparse.SUPPRESS) evidence.add_argument("--redacted", action="store_true", default=argparse.SUPPRESS)
evidence.add_argument("--include-excerpts", "--raw-ok", dest="include_excerpts", action="store_true", default=argparse.SUPPRESS) evidence.add_argument(
"--include-excerpts", "--raw-ok", dest="include_excerpts", action="store_true", default=argparse.SUPPRESS
)
edges = sub.add_parser("edges", help="Show canonical graph edges for one claim.") edges = sub.add_parser("edges", help="Show canonical graph edges for one claim.")
edges.add_argument("claim_id") edges.add_argument("claim_id")
edges.add_argument("--limit", type=int, default=12) edges.add_argument("--limit", type=int, default=12)
edges.add_argument("--format", choices=["markdown", "json"], default=argparse.SUPPRESS) edges.add_argument("--format", choices=["markdown", "json"], default=argparse.SUPPRESS)
edges.add_argument("--redacted", action="store_true", default=argparse.SUPPRESS) edges.add_argument("--redacted", action="store_true", default=argparse.SUPPRESS)
edges.add_argument("--include-excerpts", "--raw-ok", dest="include_excerpts", action="store_true", default=argparse.SUPPRESS) edges.add_argument(
"--include-excerpts", "--raw-ok", dest="include_excerpts", action="store_true", default=argparse.SUPPRESS
)
status = sub.add_parser("status", help="Read back Cloud SQL memory backend status.") status = sub.add_parser("status", help="Read back Cloud SQL memory backend status.")
status.add_argument("--format", choices=["markdown", "json"], default=argparse.SUPPRESS) status.add_argument("--format", choices=["markdown", "json"], default=argparse.SUPPRESS)
@ -96,8 +173,14 @@ def parse_args() -> argparse.Namespace:
help="Stage a review-gated canonical KB change instead of writing core truth to runtime memory.", help="Stage a review-gated canonical KB change instead of writing core truth to runtime memory.",
) )
propose.add_argument("--proposal-type", choices=["revise_claim", "revise_strategy"], required=True) propose.add_argument("--proposal-type", choices=["revise_claim", "revise_strategy"], required=True)
propose.add_argument("--target-kind", choices=["claim", "belief", "strategy", "identity", "role", "telos", "framework", "reasoning_tool", "other"], required=True) propose.add_argument(
propose.add_argument("--target-ref", required=True, help="Claim id, strategy id/name, or unresolved target description.") "--target-kind",
choices=["claim", "belief", "strategy", "identity", "role", "telos", "framework", "reasoning_tool", "other"],
required=True,
)
propose.add_argument(
"--target-ref", required=True, help="Claim id, strategy id/name, or unresolved target description."
)
propose.add_argument("--current", default="", help="Current/old claim or state, if known.") propose.add_argument("--current", default="", help="Current/old claim or state, if known.")
propose.add_argument("--proposed", required=True, help="Proposed replacement or correction.") propose.add_argument("--proposed", required=True, help="Proposed replacement or correction.")
propose.add_argument("--evidence", action="append", default=[], help="Evidence/source pointer. Can be repeated.") propose.add_argument("--evidence", action="append", default=[], help="Evidence/source pointer. Can be repeated.")
@ -124,7 +207,9 @@ def parse_args() -> argparse.Namespace:
list_proposals.add_argument("--limit", type=int, default=10) list_proposals.add_argument("--limit", type=int, default=10)
list_proposals.add_argument("--format", choices=["markdown", "json"], default=argparse.SUPPRESS) list_proposals.add_argument("--format", choices=["markdown", "json"], default=argparse.SUPPRESS)
search_proposals = sub.add_parser("search-proposals", help="Search staged KB proposals across statuses from Cloud SQL.") search_proposals = sub.add_parser(
"search-proposals", help="Search staged KB proposals across statuses from Cloud SQL."
)
search_proposals.add_argument("query") search_proposals.add_argument("query")
search_proposals.add_argument("--status", default="all") search_proposals.add_argument("--status", default="all")
search_proposals.add_argument("--limit", type=int, default=10) search_proposals.add_argument("--limit", type=int, default=10)
@ -201,7 +286,7 @@ def password(args: argparse.Namespace) -> str:
check=False, check=False,
) )
if result.returncode != 0: if result.returncode != 0:
raise SystemExit(f"Secret access failed: {result.stderr.strip()}") raise SystemExit(f"Secret access failed: {result.stderr.strip()}")
return result.stdout.strip() return result.stdout.strip()
@ -233,6 +318,157 @@ def psql_json_lines(args: argparse.Namespace, sql: str, db: str | None = None) -
return rows return rows
def canonical_json_sha256(value: Any) -> str:
encoded = json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode()
return hashlib.sha256(encoded).hexdigest()
def database_read_marker(args: argparse.Namespace) -> dict[str, Any]:
sql = """
select jsonb_build_object(
'database', current_database(),
'database_user', current_user,
'system_identifier', (select system_identifier::text from pg_control_system()),
'wal_lsn', pg_current_wal_lsn()::text
)::text;
"""
rows = psql_json_lines(args, sql, db=args.canonical_db)
if len(rows) != 1:
raise SystemExit("Canonical database read marker returned an invalid row count.")
return rows[0]
def _read_marker_stable(before: dict[str, Any], after: dict[str, Any]) -> bool:
return all(
before.get(key) == after.get(key) for key in ("database", "database_user", "system_identifier", "wal_lsn")
)
def _stable_receipt_value(value: Any) -> Any:
if isinstance(value, dict):
return {
key: _stable_receipt_value(item)
for key, item in sorted(value.items())
if key not in {"claim_url", "connected_url", "retrieval_receipt"}
}
if isinstance(value, list):
return [_stable_receipt_value(item) for item in value]
return value
def _retrieved_claim_ids(data: dict[str, Any]) -> list[str]:
values = []
if data.get("claim_id"):
values.append(str(data["claim_id"]))
claim = data.get("claim") or {}
if claim.get("id"):
values.append(str(claim["id"]))
values.extend(str(row["id"]) for row in data.get("claims", []) if row.get("id"))
return sorted(set(values))
def _retrieved_evidence(data: dict[str, Any]) -> list[dict[str, Any]]:
rows = [row for row in data.get("evidence", []) if isinstance(row, dict)]
for claim in data.get("claims", []):
rows.extend(row for row in claim.get("evidence", []) if isinstance(row, dict))
return rows
def build_retrieval_receipt(
data: dict[str, Any],
*,
query_descriptor: dict[str, Any],
before: dict[str, Any],
after: dict[str, Any],
attempts: int,
consistency_status: str,
) -> dict[str, Any]:
evidence = _retrieved_evidence(data)
artifact_states = sorted(
(
{
"source_id": row.get("source_id"),
"source_hash": row.get("source_hash"),
"storage_path": row.get("storage_path"),
}
for row in evidence
),
key=lambda row: (str(row.get("source_id") or ""), str(row.get("storage_path") or "")),
)
return {
"schema": RETRIEVAL_RECEIPT_SCHEMA,
"query_sha256": canonical_json_sha256(query_descriptor),
"semantic_context_sha256": canonical_json_sha256(_stable_receipt_value(data)),
"artifact_state_sha256": canonical_json_sha256(artifact_states),
"claim_ids": _retrieved_claim_ids(data),
"source_ids": sorted({str(row["source_id"]) for row in evidence if row.get("source_id")}),
"counts": {
"claims": len(_retrieved_claim_ids(data)),
"evidence_rows": len(evidence),
"edge_rows": len(data.get("edges", []))
+ sum(len(claim.get("edges", [])) for claim in data.get("claims", [])),
},
"read_consistency": {
"status": consistency_status,
"attempts": attempts,
"database": before.get("database"),
"database_user": before.get("database_user"),
"system_identifier": before.get("system_identifier"),
"wal_lsn_before": before.get("wal_lsn"),
"wal_lsn_after": after.get("wal_lsn"),
},
}
def receipt_query_descriptor(args: argparse.Namespace) -> dict[str, Any]:
return {
"command": args.command,
"query": getattr(args, "query", None),
"claim_id": getattr(args, "claim_id", None),
"limit": getattr(args, "limit", None),
"context_limit": getattr(args, "context_limit", None),
}
def read_with_retrieval_receipt(args: argparse.Namespace, loader: Any) -> dict[str, Any]:
try:
max_attempts = max(1, min(int(os.environ.get("TELEO_KB_READ_ATTEMPTS", "3")), 5))
except ValueError as exc:
raise SystemExit("TELEO_KB_READ_ATTEMPTS must be an integer") from exc
previous_semantic_sha256: str | None = None
last_markers: tuple[dict[str, Any], dict[str, Any]] | None = None
descriptor = receipt_query_descriptor(args)
for attempt in range(1, max_attempts + 1):
before = database_read_marker(args)
data = loader()
after = database_read_marker(args)
last_markers = (before, after)
consistency_status = "stable_wal_marker" if _read_marker_stable(before, after) else "wal_changed_during_read"
receipt = build_retrieval_receipt(
data,
query_descriptor=descriptor,
before=before,
after=after,
attempts=attempt,
consistency_status=consistency_status,
)
if consistency_status == "stable_wal_marker":
data["retrieval_receipt"] = receipt
return data
semantic_sha256 = receipt["semantic_context_sha256"]
if previous_semantic_sha256 == semantic_sha256:
receipt["read_consistency"]["status"] = "stable_content_across_wal_change_retry"
data["retrieval_receipt"] = receipt
return data
previous_semantic_sha256 = semantic_sha256
assert last_markers is not None
before, after = last_markers
raise SystemExit(
"Canonical DB changed during retrieval and no identical retry was observed "
f"(wal_lsn {before.get('wal_lsn')} -> {after.get('wal_lsn')})."
)
def include_excerpts(args: argparse.Namespace) -> bool: def include_excerpts(args: argparse.Namespace) -> bool:
return bool(getattr(args, "include_excerpts", False)) and not bool(getattr(args, "redacted", False)) return bool(getattr(args, "include_excerpts", False)) and not bool(getattr(args, "redacted", False))
@ -394,11 +630,13 @@ select jsonb_build_object(
evidence_sql = f""" evidence_sql = f"""
with ranked as ( with ranked as (
select ce.claim_id, select ce.claim_id,
s.id as source_id,
ce.role::text as role, ce.role::text as role,
ce.weight, ce.weight,
s.source_type, s.source_type,
s.url, s.url,
s.storage_path, s.storage_path,
s.hash as source_hash,
left(coalesce(s.excerpt, ''), 800) as excerpt, left(coalesce(s.excerpt, ''), 800) as excerpt,
row_number() over ( row_number() over (
partition by ce.claim_id partition by ce.claim_id
@ -413,11 +651,13 @@ with ranked as (
) )
select jsonb_build_object( select jsonb_build_object(
'claim_id', claim_id::text, 'claim_id', claim_id::text,
'source_id', source_id::text,
'role', role, 'role', role,
'weight', weight, 'weight', weight,
'source_type', source_type, 'source_type', source_type,
'url', url, 'url', url,
'storage_path', storage_path, 'storage_path', storage_path,
'source_hash', source_hash,
'excerpt', excerpt 'excerpt', excerpt
)::text )::text
from ranked from ranked
@ -501,7 +741,8 @@ select jsonb_build_object(
"excerpt": claim.get("text"), "excerpt": claim.get("text"),
} }
for claim in claims for claim in claims
] + [ ]
+ [
{ {
"source_table": f"public.{row['source']}", "source_table": f"public.{row['source']}",
"row_id": f"{row['owner']}:{row['title']}", "row_id": f"{row['owner']}:{row['title']}",
@ -524,9 +765,13 @@ select jsonb_build_object(
row.pop("body", None) row.pop("body", None)
for hit in result["hits"]: for hit in result["hits"]:
hit.pop("excerpt", None) hit.pop("excerpt", None)
result["privacy"] = "redacted proof mode: raw body used only inside SQL matching; output has counts, hashes, canonical table, row id, timestamps, and evidence links" result["privacy"] = (
"redacted proof mode: raw body used only inside SQL matching; output has counts, hashes, canonical table, row id, timestamps, and evidence links"
)
else: else:
result["privacy"] = "private agent context mode: short excerpts included; do not retain or paste raw output into public artifacts" result["privacy"] = (
"private agent context mode: short excerpts included; do not retain or paste raw output into public artifacts"
)
return result return result
@ -609,9 +854,13 @@ select json_build_object(
if not include_excerpts(args): if not include_excerpts(args):
for hit in result["hits"]: for hit in result["hits"]:
hit.pop("excerpt", None) hit.pop("excerpt", None)
result["privacy"] = "redacted proof mode: raw body used only inside SQL matching; output has counts, hashes, source table, row id, and timestamps" result["privacy"] = (
"redacted proof mode: raw body used only inside SQL matching; output has counts, hashes, source table, row id, and timestamps"
)
else: else:
result["privacy"] = "private agent context mode: short excerpts included; do not retain or paste raw output into public artifacts" result["privacy"] = (
"private agent context mode: short excerpts included; do not retain or paste raw output into public artifacts"
)
return result return result
@ -619,6 +868,11 @@ def query_rows(args: argparse.Namespace, query: str, limit: int) -> dict[str, An
canonical = query_canonical_rows(args, query, limit) canonical = query_canonical_rows(args, query, limit)
if canonical.get("hit_count_total", 0) or canonical.get("hits"): if canonical.get("hit_count_total", 0) or canonical.get("hits"):
return canonical return canonical
if not audit_restore_available(args):
canonical["canonical_fallback_reason"] = (
"no matching canonical public/persona/strategy/belief rows; teleo_restore audit fallback unavailable"
)
return canonical
audit = query_audit_rows(args, query, limit) audit = query_audit_rows(args, query, limit)
audit["canonical_fallback_reason"] = "no matching canonical public/persona/strategy/belief rows" audit["canonical_fallback_reason"] = "no matching canonical public/persona/strategy/belief rows"
return audit return audit
@ -652,11 +906,13 @@ def canonical_evidence(args: argparse.Namespace, claim_id: str, limit: int) -> l
sql = f""" sql = f"""
with ranked as ( with ranked as (
select ce.claim_id, select ce.claim_id,
s.id as source_id,
ce.role::text as role, ce.role::text as role,
ce.weight, ce.weight,
s.source_type, s.source_type,
s.url, s.url,
s.storage_path, s.storage_path,
s.hash as source_hash,
left(coalesce(s.excerpt, ''), 800) as excerpt, left(coalesce(s.excerpt, ''), 800) as excerpt,
row_number() over ( row_number() over (
partition by ce.claim_id partition by ce.claim_id
@ -671,11 +927,13 @@ with ranked as (
) )
select jsonb_build_object( select jsonb_build_object(
'claim_id', claim_id::text, 'claim_id', claim_id::text,
'source_id', source_id::text,
'role', role, 'role', role,
'weight', weight, 'weight', weight,
'source_type', source_type, 'source_type', source_type,
'url', url, 'url', url,
'storage_path', storage_path, 'storage_path', storage_path,
'source_hash', source_hash,
'excerpt', excerpt 'excerpt', excerpt
)::text )::text
from ranked from ranked
@ -1106,6 +1364,23 @@ select jsonb_build_object(
return result return result
def audit_restore_available(args: argparse.Namespace) -> bool:
sql = """
select bool_and(to_regclass(table_name) is not null)
from (values
('teleo_restore.response_audit'),
('teleo_restore.sources'),
('teleo_restore.agent_research_runs'),
('teleo_restore.agent_tool_invocations'),
('teleo_restore.audit_log'),
('teleo_restore.metrics_snapshots'),
('teleo_restore.prs'),
('teleo_restore.review_records')
) required(table_name);
"""
return run_psql(args, sql).strip().lower() in {"t", "true"}
def status(args: argparse.Namespace) -> dict[str, Any]: def status(args: argparse.Namespace) -> dict[str, Any]:
canonical_sql = """ canonical_sql = """
select json_build_object( select json_build_object(
@ -1158,10 +1433,24 @@ select json_build_object(
) )
)::text; )::text;
""" """
result = json.loads(run_psql(args, audit_sql).strip()) canonical = json.loads(run_psql(args, canonical_sql, db=args.canonical_db).strip())
result["canonical"] = json.loads(run_psql(args, canonical_sql, db=args.canonical_db).strip()) audit_available = audit_restore_available(args)
if audit_available:
result = json.loads(run_psql(args, audit_sql).strip())
else:
result = {
"db_identity": canonical["db_identity"],
"extensions": canonical.get("extensions") or [],
"teleo_restore_tables": 0,
"total_rows": 0,
"high_signal_rows": {},
}
result["canonical"] = canonical
result["audit_fallback_available"] = audit_available
result["artifact"] = "teleo_cloudsql_memory_status" result["artifact"] = "teleo_cloudsql_memory_status"
result["backend"] = f"cloudsql:teleo-pgvector-standby/{args.canonical_db}/public+kb_stage primary; {args.db}/teleo_restore fallback" result["backend"] = f"cloudsql:teleo-pgvector-standby/{args.canonical_db}/public+kb_stage primary; " + (
f"{args.db}/teleo_restore fallback" if audit_available else "teleo_restore fallback unavailable"
)
return result return result
@ -1170,6 +1459,15 @@ def emit_json(value: dict[str, Any]) -> None:
def emit_markdown(value: dict[str, Any]) -> None: def emit_markdown(value: dict[str, Any]) -> None:
receipt = value.get("retrieval_receipt") or {}
if receipt:
consistency = receipt.get("read_consistency") or {}
print("# Teleo KB Retrieval Receipt\n")
print(f"- schema: `{receipt.get('schema')}`")
print(f"- semantic context SHA-256: `{receipt.get('semantic_context_sha256')}`")
print(f"- artifact state SHA-256: `{receipt.get('artifact_state_sha256')}`")
print(f"- DB read consistency: `{consistency.get('status')}`; attempts: `{consistency.get('attempts')}`\n")
if value["artifact"] == "teleo_cloudsql_memory_status": if value["artifact"] == "teleo_cloudsql_memory_status":
print("# Teleo Cloud SQL Memory Status\n") print("# Teleo Cloud SQL Memory Status\n")
print(f"- Backend: `{value['backend']}`") print(f"- Backend: `{value['backend']}`")
@ -1180,10 +1478,7 @@ def emit_markdown(value: dict[str, Any]) -> None:
print(f"- Canonical tables: `{json.dumps(canonical.get('schema_tables'), sort_keys=True)}`") print(f"- Canonical tables: `{json.dumps(canonical.get('schema_tables'), sort_keys=True)}`")
print(f"- Canonical high-signal rows: `{json.dumps(canonical.get('high_signal_rows'), sort_keys=True)}`") print(f"- Canonical high-signal rows: `{json.dumps(canonical.get('high_signal_rows'), sort_keys=True)}`")
counts = canonical.get("high_signal_rows") or {} counts = canonical.get("high_signal_rows") or {}
if all( if all(key in counts for key in ("claims", "sources", "claim_edges", "claim_evidence", "kb_proposals")):
key in counts
for key in ("claims", "sources", "claim_edges", "claim_evidence", "kb_proposals")
):
print( print(
f"\nDB readback: claims: `{counts['claims']}`; sources: `{counts['sources']}`; " f"\nDB readback: claims: `{counts['claims']}`; sources: `{counts['sources']}`; "
f"claim_edges: `{counts['claim_edges']}`; claim_evidence: `{counts['claim_evidence']}`; " f"claim_edges: `{counts['claim_edges']}`; claim_evidence: `{counts['claim_evidence']}`; "
@ -1218,7 +1513,9 @@ def emit_markdown(value: dict[str, Any]) -> None:
print(f"### {i}. {markdown_claim_link(claim['id'], claim_label[:140])}\n") print(f"### {i}. {markdown_claim_link(claim['id'], claim_label[:140])}\n")
print(f"- claim id: {markdown_claim_link(claim['id'], claim['id'])}") print(f"- claim id: {markdown_claim_link(claim['id'], claim['id'])}")
print(f"- open full claim/body/edges: {markdown_claim_link(claim['id'], 'claim page')}") print(f"- open full claim/body/edges: {markdown_claim_link(claim['id'], 'claim page')}")
print(f"- type: `{claim.get('type')}`; confidence: `{claim.get('confidence')}`; score: `{claim.get('score')}`") print(
f"- type: `{claim.get('type')}`; confidence: `{claim.get('confidence')}`; score: `{claim.get('score')}`"
)
print(f"- tags: `{', '.join(claim.get('tags') or [])}`") print(f"- tags: `{', '.join(claim.get('tags') or [])}`")
print(f"- evidence rows: `{claim.get('evidence_count', len(claim.get('evidence', [])))}`") print(f"- evidence rows: `{claim.get('evidence_count', len(claim.get('evidence', [])))}`")
print(f"- edge rows: `{claim.get('edge_count', len(claim.get('edges', [])))}`") print(f"- edge rows: `{claim.get('edge_count', len(claim.get('edges', [])))}`")
@ -1246,7 +1543,9 @@ def emit_markdown(value: dict[str, Any]) -> None:
print(f"- Backend: `{value['backend']}`") print(f"- Backend: `{value['backend']}`")
print(f"- claim id: {markdown_claim_link(claim['id'], claim['id'])}") print(f"- claim id: {markdown_claim_link(claim['id'], claim['id'])}")
print(f"- open full claim/body/edges: {markdown_claim_link(claim['id'], 'claim page')}") print(f"- open full claim/body/edges: {markdown_claim_link(claim['id'], 'claim page')}")
print(f"- type: `{claim.get('type')}`; status: `{claim.get('status')}`; confidence: `{claim.get('confidence')}`") print(
f"- type: `{claim.get('type')}`; status: `{claim.get('status')}`; confidence: `{claim.get('confidence')}`"
)
print(f"- tags: `{', '.join(claim.get('tags') or [])}`") print(f"- tags: `{', '.join(claim.get('tags') or [])}`")
if claim.get("text"): if claim.get("text"):
print(f"\n{markdown_claim_link(claim['id'], claim['text'][:180])}\n") print(f"\n{markdown_claim_link(claim['id'], claim['text'][:180])}\n")
@ -1276,7 +1575,11 @@ def emit_markdown(value: dict[str, Any]) -> None:
return return
if value["artifact"] in {"teleo_cloudsql_kb_core_change_proposal", "teleo_cloudsql_kb_edge_proposal"}: if value["artifact"] in {"teleo_cloudsql_kb_core_change_proposal", "teleo_cloudsql_kb_edge_proposal"}:
title = "Staged KB Edge Proposal" if value["artifact"] == "teleo_cloudsql_kb_edge_proposal" else "Staged KB Core-Change Proposal" title = (
"Staged KB Edge Proposal"
if value["artifact"] == "teleo_cloudsql_kb_edge_proposal"
else "Staged KB Core-Change Proposal"
)
print(f"# {title}\n") print(f"# {title}\n")
print(f"- Proposal id: `{value['id']}`") print(f"- Proposal id: `{value['id']}`")
print(f"- Type: `{value['proposal_type']}`") print(f"- Type: `{value['proposal_type']}`")
@ -1372,28 +1675,26 @@ def emit_markdown(value: dict[str, Any]) -> None:
def main() -> None: def main() -> None:
args = parse_args() args = parse_args()
if args.command == "status": read_loaders = {
result = status(args) "search": lambda: query_rows(args, args.query, args.limit),
elif args.command == "show": "context": lambda: query_rows(args, args.query, args.limit),
result = show_canonical_claim(args) "show": lambda: show_canonical_claim(args),
elif args.command == "evidence": "evidence": lambda: evidence_canonical_claim(args),
result = evidence_canonical_claim(args) "edges": lambda: edges_canonical_claim(args),
elif args.command == "edges": "status": lambda: status(args),
result = edges_canonical_claim(args) "list-proposals": lambda: list_proposals(args),
"search-proposals": lambda: search_proposals(args),
"show-proposal": lambda: show_proposal(args),
"decision-matrix-status": lambda: decision_matrix_status(args),
}
if args.command in read_loaders:
result = read_with_retrieval_receipt(args, read_loaders[args.command])
elif args.command == "propose-core-change": elif args.command == "propose-core-change":
result = propose_core_change(args) result = propose_core_change(args)
elif args.command == "propose-edge": elif args.command == "propose-edge":
result = propose_edge(args) result = propose_edge(args)
elif args.command == "list-proposals":
result = list_proposals(args)
elif args.command == "search-proposals":
result = search_proposals(args)
elif args.command == "show-proposal":
result = show_proposal(args)
elif args.command == "decision-matrix-status":
result = decision_matrix_status(args)
else: else:
result = query_rows(args, args.query, args.limit) raise SystemExit(f"Unsupported command: {args.command}")
if args.format == "json": if args.format == "json":
emit_json(result) emit_json(result)
else: else:

View file

@ -0,0 +1,717 @@
#!/usr/bin/env python3
"""Restore a VPS canonical snapshot into one disposable private Cloud SQL database.
The helper is intentionally narrower than a production cutover tool. It can
create only a previously absent ``teleo_clone_*`` database, verifies full
manifest parity, and deletes the clone automatically whenever restore or proof
fails. A successful clone is retained only for a bounded no-send replay and is
removed with the explicit ``cleanup`` operation afterward.
"""
from __future__ import annotations
import argparse
import hashlib
import ipaddress
import json
import os
import re
import signal
import stat
import subprocess
import time
import uuid
from datetime import UTC, datetime
from pathlib import Path
from typing import Any
try:
from .verify_postgres_parity_manifest import compare_manifests, load_manifest
except ImportError: # pragma: no cover - direct script execution on the GCP VM
from verify_postgres_parity_manifest import compare_manifests, load_manifest
DEFAULT_HOST = "10.61.0.3"
DEFAULT_PROJECT = "teleo-501523"
DEFAULT_SECRET = "gcp-teleo-pgvector-standby-postgres-password"
DEFAULT_SERVICE = "leoclean-gcp-prod-parallel.service"
DEFAULT_RUN_ROOT = Path("/var/lib/teleo-gcp-restore-runs")
DEFAULT_MANIFEST_SQL = Path(__file__).with_name("postgres_parity_manifest.sql")
REVIEWED_MANIFEST_SQL_SHA256 = "8b8cdc25d54fdd8de05eb38c6e4423d2836953eb6012d4545f5c9c71b5f0150a"
TARGET_DATABASE_RE = re.compile(r"teleo_clone_[a-z0-9][a-z0-9_]{0,50}\Z")
RUN_ID_RE = re.compile(r"gcp-restore-[a-z0-9][a-z0-9-]{7,47}\Z")
SHA256_RE = re.compile(r"[0-9a-f]{64}\Z")
LOCALE_RE = re.compile(r"[A-Za-z0-9_.@-]{1,64}\Z")
PG_RESTORE_VERSION_RE = re.compile(r"pg_restore \(PostgreSQL\) (?P<major>[0-9]+)(?:\.[0-9]+)*")
ROLLBACK_DATABASE = "teleo_canonical_pre_20260712t1905z"
class RestoreError(RuntimeError):
"""Raised when a guarded restore or cleanup invariant fails."""
class TerminationRequested(RestoreError):
"""Raised after SIGINT/SIGTERM so clone cleanup still runs."""
def utc_now() -> str:
return datetime.now(UTC).isoformat()
def sha256_file(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as handle:
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
digest.update(chunk)
return digest.hexdigest()
def validate_regular_file(path: Path, label: str) -> Path:
try:
mode = path.lstat().st_mode
except OSError as exc:
raise RestoreError(f"{label} is unavailable: {exc}") from exc
if not stat.S_ISREG(mode) or path.is_symlink():
raise RestoreError(f"{label} must be a regular non-symlink file")
return path.resolve()
def validate_custom_dump(path: Path, expected_sha256: str) -> dict[str, Any]:
path = validate_regular_file(path, "custom dump")
if path.stat().st_size <= 5:
raise RestoreError("custom dump is empty or truncated")
with path.open("rb") as handle:
if handle.read(5) != b"PGDMP":
raise RestoreError("custom dump is missing the PGDMP signature")
actual_sha256 = sha256_file(path)
if actual_sha256 != expected_sha256:
raise RestoreError("custom dump SHA-256 does not match --expected-dump-sha256")
return {"path": str(path), "bytes": path.stat().st_size, "sha256": actual_sha256}
def _run(
command: list[str],
*,
timeout: float,
env: dict[str, str] | None = None,
input_text: str | None = None,
) -> subprocess.CompletedProcess[str]:
return subprocess.run(
command,
capture_output=True,
check=False,
env=env,
input=input_text,
text=True,
timeout=timeout,
)
def _bounded_error(completed: subprocess.CompletedProcess[str], secrets: tuple[str, ...] = ()) -> str:
message = (completed.stderr or completed.stdout or "no command output").strip()
for secret in secrets:
if secret:
message = message.replace(secret, "[REDACTED]")
return message[-2000:]
def _require_success(
completed: subprocess.CompletedProcess[str],
action: str,
*,
secrets: tuple[str, ...] = (),
) -> str:
if completed.returncode != 0:
raise RestoreError(f"{action} failed (exit {completed.returncode}): {_bounded_error(completed, secrets)}")
return completed.stdout
def service_state(service: str, *, timeout: float) -> dict[str, Any]:
completed = _run(
[
"systemctl",
"show",
service,
"-p",
"ActiveState",
"-p",
"SubState",
"-p",
"MainPID",
"-p",
"NRestarts",
"-p",
"ExecMainStartTimestamp",
"--no-pager",
],
timeout=timeout,
)
raw = _require_success(completed, "live GCP service readback")
result = {}
for line in raw.splitlines():
key, separator, value = line.partition("=")
if separator:
result[key] = value
required = {"ActiveState", "SubState", "MainPID", "NRestarts", "ExecMainStartTimestamp"}
if required - set(result):
raise RestoreError("live GCP service readback omitted required fields")
return result
def resolve_password(project: str, secret: str, *, timeout: float) -> str:
completed = _run(
[
"gcloud",
"secrets",
"versions",
"access",
"latest",
f"--secret={secret}",
f"--project={project}",
"--quiet",
],
timeout=timeout,
)
password = _require_success(completed, "Cloud SQL credential resolution").strip()
if not password:
raise RestoreError("Cloud SQL credential resolved to an empty value")
return password
def validate_pg_restore_version(
pg_restore_bin: str,
source_manifest: dict[str, Any],
*,
timeout: float,
) -> dict[str, Any]:
completed = _run([pg_restore_bin, "--version"], timeout=timeout)
raw = _require_success(completed, "pg_restore version preflight").strip()
match = PG_RESTORE_VERSION_RE.search(raw)
if match is None:
raise RestoreError("pg_restore version preflight returned an unrecognized value")
client_major = int(match.group("major"))
source_version_num = int(source_manifest["singleton"]["identity"]["server_version_num"])
source_major = source_version_num // 10000
if client_major < source_major:
raise RestoreError(f"pg_restore major {client_major} is older than source PostgreSQL major {source_major}")
return {
"pg_restore_bin": pg_restore_bin,
"pg_restore_version": raw,
"pg_restore_major": client_major,
"source_postgres_major": source_major,
"compatible": True,
}
def postgres_env(password: str, *, read_only: bool) -> dict[str, str]:
env = {**os.environ, "PGPASSWORD": password}
options = ["-c statement_timeout=120000"]
if read_only:
options.append("-c default_transaction_read_only=on")
env["PGOPTIONS"] = " ".join(options)
return env
def psql_command(args: argparse.Namespace, database: str) -> list[str]:
return [
args.psql_bin,
f"host={args.host} port=5432 dbname={database} user=postgres sslmode=require connect_timeout=8",
"-X",
"-Atq",
"-v",
"ON_ERROR_STOP=1",
]
def run_psql(
args: argparse.Namespace,
password: str,
database: str,
sql: str,
*,
read_only: bool,
timeout: float,
) -> str:
completed = _run(
psql_command(args, database),
timeout=timeout,
env=postgres_env(password, read_only=read_only),
input_text=sql,
)
return _require_success(completed, "PostgreSQL command", secrets=(password,))
def database_exists(args: argparse.Namespace, password: str, database: str) -> bool:
output = run_psql(
args,
password,
"postgres",
f"select count(*) from pg_database where datname = '{database}';\n",
read_only=True,
timeout=args.command_timeout,
).strip()
if output not in {"0", "1"}:
raise RestoreError("database existence readback was invalid")
return output == "1"
def create_database(
args: argparse.Namespace,
password: str,
source_manifest: dict[str, Any],
) -> None:
identity = source_manifest["singleton"]["identity"]
encoding = str(identity.get("server_encoding") or "")
collation = str(identity.get("database_collation") or "")
ctype = str(identity.get("database_ctype") or "")
if encoding != "UTF8":
raise RestoreError(f"source encoding is not the reviewed UTF8 value: {encoding!r}")
if not LOCALE_RE.fullmatch(collation) or not LOCALE_RE.fullmatch(ctype):
raise RestoreError("source collation or ctype is not a safe locale identifier")
sql = (
f'create database "{args.target_db}" with template template0 '
f"encoding 'UTF8' lc_collate '{collation}' lc_ctype '{ctype}';\n"
)
run_psql(
args,
password,
"postgres",
sql,
read_only=False,
timeout=args.command_timeout,
)
def restore_dump(args: argparse.Namespace, password: str) -> None:
command = [
args.pg_restore_bin,
"-d",
f"host={args.host} port=5432 dbname={args.target_db} user=postgres sslmode=require connect_timeout=8",
"--no-owner",
"--no-acl",
"--exit-on-error",
str(args.dump.resolve()),
]
completed = _run(
command,
timeout=args.restore_timeout,
env=postgres_env(password, read_only=False),
)
_require_success(completed, "canonical PostgreSQL restore", secrets=(password,))
def capture_target_manifest(args: argparse.Namespace, password: str) -> str:
command = [*psql_command(args, args.target_db), "-f", str(args.manifest_sql.resolve())]
completed = _run(
command,
timeout=args.manifest_timeout,
env=postgres_env(password, read_only=True),
)
return _require_success(completed, "target parity manifest capture", secrets=(password,))
def validate_private_identity(target_manifest: dict[str, Any], target_db: str) -> dict[str, Any]:
identity = target_manifest["singleton"]["identity"]
if identity.get("database") != target_db:
raise RestoreError("target manifest database does not match the generated clone")
if identity.get("ssl") is not True:
raise RestoreError("target manifest did not prove TLS")
if identity.get("transaction_read_only") != "on":
raise RestoreError("target manifest was not captured in a read-only transaction")
try:
address = ipaddress.ip_address(str(identity.get("server_address") or ""))
except ValueError as exc:
raise RestoreError("target manifest server address is invalid") from exc
private = any(
address in network
for network in (
ipaddress.ip_network("10.0.0.0/8"),
ipaddress.ip_network("172.16.0.0/12"),
ipaddress.ip_network("192.168.0.0/16"),
)
)
if not private:
raise RestoreError("target manifest server address is not RFC1918-private")
return {
"target_database": target_db,
"server_address": str(address),
"server_port": identity.get("server_port"),
"ssl": True,
"ssl_version": identity.get("ssl_version"),
"private_connectivity": True,
"transaction_read_only": "on",
"source_compute": "teleo-prod-1",
}
def drop_clone(args: argparse.Namespace, password: str) -> dict[str, Any]:
if not TARGET_DATABASE_RE.fullmatch(args.target_db):
raise RestoreError("refusing to drop a database outside teleo_clone_*")
existed_before = database_exists(args, password, args.target_db)
if existed_before:
sql = f"""
select pg_terminate_backend(pid)
from pg_stat_activity
where datname = '{args.target_db}' and pid <> pg_backend_pid();
drop database "{args.target_db}";
"""
run_psql(
args,
password,
"postgres",
sql,
read_only=False,
timeout=args.command_timeout,
)
remaining = 1 if database_exists(args, password, args.target_db) else 0
if remaining:
raise RestoreError("generated clone remained after cleanup")
return {"existed_before": existed_before, "clone_database_remaining": remaining}
def rollback_database_state(args: argparse.Namespace, password: str) -> dict[str, Any]:
output = run_psql(
args,
password,
"postgres",
f"""
select jsonb_build_object(
'exists', (select count(*) = 1 from pg_database where datname = '{ROLLBACK_DATABASE}'),
'datallowconn', (select datallowconn from pg_database where datname = '{ROLLBACK_DATABASE}'),
'connections', (select count(*) from pg_stat_activity where datname = '{ROLLBACK_DATABASE}')
)::text;
""",
read_only=True,
timeout=args.command_timeout,
)
rows = [line for line in output.splitlines() if line.strip().startswith("{")]
if len(rows) != 1:
raise RestoreError("rollback database state readback was invalid")
state = json.loads(rows[0])
if state != {"exists": True, "datallowconn": False, "connections": 0}:
raise RestoreError("disabled rollback database invariant changed")
return state
def secure_run_dir(run_root: Path, run_id: str, *, create: bool) -> Path:
if run_root.is_symlink():
raise RestoreError("run root must not be a symlink")
root = run_root.resolve()
if create:
root.mkdir(parents=True, exist_ok=True, mode=0o700)
os.chmod(root, 0o700)
if not root.is_dir() or root.is_symlink():
raise RestoreError("run root is absent or unsafe")
run_dir = root / run_id
if create:
try:
run_dir.mkdir(mode=0o700)
except FileExistsError as exc:
raise RestoreError("run directory already exists") from exc
if not run_dir.is_dir() or run_dir.is_symlink() or run_dir.resolve().parent != root:
raise RestoreError("run directory escaped the fixed root or is unsafe")
os.chmod(run_dir, 0o700)
return run_dir
def write_private(path: Path, content: str | bytes) -> None:
encoded = content.encode() if isinstance(content, str) else content
temporary = path.parent / f".{path.name}.{uuid.uuid4().hex}.tmp"
descriptor = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
try:
with os.fdopen(descriptor, "wb") as handle:
handle.write(encoded)
handle.flush()
os.fsync(handle.fileno())
os.replace(temporary, path)
os.chmod(path, 0o600)
finally:
try:
temporary.unlink()
except FileNotFoundError:
pass
def run_restore(args: argparse.Namespace) -> dict[str, Any]:
started = time.monotonic()
run_dir = secure_run_dir(args.run_root, args.run_id, create=True)
receipt: dict[str, Any] = {
"artifact": "gcp_generated_postgres_snapshot_restore",
"generated_at_utc": utc_now(),
"run_id": args.run_id,
"target_database": args.target_db,
"status": "running",
"phase": "validation",
"source": {},
"toolchain": {},
"target": {},
"parity": {},
"live_service": {},
"rollback_database": {},
"cleanup": {"required": True, "attempted": False, "clone_database_remaining": None},
"safety": {
"live_database_named": False,
"live_profile_modified": False,
"live_service_restarted": False,
"telegram_message_sent": False,
"secret_persisted": False,
},
"error": None,
}
password = ""
creation_attempted = False
try:
if os.geteuid() != 0:
raise RestoreError("restore must run as root on the GCP staging VM")
args.dump = validate_regular_file(args.dump, "custom dump")
args.source_manifest = validate_regular_file(args.source_manifest, "source manifest")
args.manifest_sql = validate_regular_file(args.manifest_sql, "parity manifest SQL")
if sha256_file(args.manifest_sql) != REVIEWED_MANIFEST_SQL_SHA256:
raise RestoreError("parity manifest SQL does not match the reviewed SHA-256")
receipt["source"]["dump"] = validate_custom_dump(args.dump, args.expected_dump_sha256)
receipt["source"]["manifest"] = {
"path": str(args.source_manifest),
"bytes": args.source_manifest.stat().st_size,
"sha256": sha256_file(args.source_manifest),
}
source_manifest = load_manifest(args.source_manifest)
receipt["toolchain"] = validate_pg_restore_version(
args.pg_restore_bin,
source_manifest,
timeout=args.command_timeout,
)
receipt["live_service"]["before"] = service_state(args.service, timeout=args.command_timeout)
receipt["phase"] = "credential_resolution"
password = resolve_password(args.project, args.password_secret, timeout=args.command_timeout)
if database_exists(args, password, args.target_db):
raise RestoreError("target clone already exists; refusing to overwrite it")
receipt["phase"] = "database_create"
creation_attempted = True
create_database(args, password, source_manifest)
receipt["phase"] = "pg_restore"
restore_started = time.monotonic()
restore_dump(args, password)
receipt["target"]["restore_seconds"] = round(time.monotonic() - restore_started, 6)
receipt["phase"] = "target_manifest"
target_manifest_text = capture_target_manifest(args, password)
target_manifest_path = run_dir / "target-manifest.jsonl"
write_private(target_manifest_path, target_manifest_text)
target_manifest = load_manifest(target_manifest_path)
receipt["target"].update(
{
"manifest_path": str(target_manifest_path),
"manifest_bytes": target_manifest_path.stat().st_size,
"manifest_sha256": sha256_file(target_manifest_path),
"connectivity": validate_private_identity(target_manifest, args.target_db),
}
)
receipt["phase"] = "parity_compare"
problems, details = compare_manifests(
source_manifest,
target_manifest,
max_target_query_ms=args.max_target_query_ms,
max_target_source_ratio=args.max_target_source_ratio,
)
receipt["parity"] = {
"status": "pass" if not problems else "fail",
"problems": problems,
"details": details,
"verifier": "ops.verify_postgres_parity_manifest.compare_manifests",
}
if problems:
raise RestoreError("canonical source/target parity comparison failed")
receipt["phase"] = "service_and_rollback_readback"
receipt["rollback_database"] = rollback_database_state(args, password)
receipt["live_service"]["after"] = service_state(args.service, timeout=args.command_timeout)
receipt["live_service"]["unchanged"] = receipt["live_service"]["before"] == receipt["live_service"]["after"]
if not receipt["live_service"]["unchanged"]:
raise RestoreError("live GCP service state changed during disposable restore")
receipt["status"] = "pass"
receipt["phase"] = "retained_for_no_send_replay"
receipt["cleanup"] = {
"required": True,
"attempted": False,
"clone_database_remaining": 1,
"exact_command": (
f"sudo python3 ops/restore_gcp_generated_postgres_snapshot.py cleanup --execute "
f"--target-db {args.target_db} --run-id {args.run_id}"
),
}
except (
OSError,
ValueError,
KeyError,
TypeError,
json.JSONDecodeError,
subprocess.TimeoutExpired,
RestoreError,
KeyboardInterrupt,
) as exc:
receipt["status"] = "fail"
receipt["error"] = {
"phase": receipt.get("phase"),
"type": type(exc).__name__,
"message": str(exc)[-2000:],
}
if creation_attempted and password:
receipt["cleanup"]["attempted"] = True
try:
receipt["cleanup"].update(drop_clone(args, password))
except Exception as cleanup_exc: # preserve both failures in the receipt
receipt["cleanup"]["error"] = str(cleanup_exc)[-2000:]
elif password:
try:
receipt["cleanup"]["clone_database_remaining"] = (
1 if database_exists(args, password, args.target_db) else 0
)
except Exception as cleanup_readback_exc:
receipt["cleanup"]["readback_error"] = str(cleanup_readback_exc)[-2000:]
finally:
if "after" not in receipt["live_service"] and "before" in receipt["live_service"]:
try:
receipt["live_service"]["after"] = service_state(args.service, timeout=args.command_timeout)
receipt["live_service"]["unchanged"] = (
receipt["live_service"]["before"] == receipt["live_service"]["after"]
)
except Exception as service_exc:
receipt["live_service"]["after_error"] = str(service_exc)[-2000:]
receipt["duration_seconds"] = round(time.monotonic() - started, 6)
write_private(run_dir / "restore-receipt.json", json.dumps(receipt, indent=2, sort_keys=True) + "\n")
password = ""
return receipt
def run_cleanup(args: argparse.Namespace) -> dict[str, Any]:
run_dir = secure_run_dir(args.run_root, args.run_id, create=False)
restore_receipt_path = run_dir / "restore-receipt.json"
restore_receipt_path = validate_regular_file(restore_receipt_path, "restore receipt")
restore_receipt = json.loads(restore_receipt_path.read_text(encoding="utf-8"))
if restore_receipt.get("status") != "pass":
raise RestoreError("cleanup requires a passing restore receipt")
if restore_receipt.get("target_database") != args.target_db:
raise RestoreError("cleanup target does not match the retained restore receipt")
if restore_receipt.get("run_id") != args.run_id:
raise RestoreError("cleanup run id does not match the retained restore receipt")
before = service_state(args.service, timeout=args.command_timeout)
password = resolve_password(args.project, args.password_secret, timeout=args.command_timeout)
try:
dropped = drop_clone(args, password)
rollback = rollback_database_state(args, password)
finally:
password = ""
after = service_state(args.service, timeout=args.command_timeout)
payload = {
"artifact": "gcp_generated_postgres_snapshot_cleanup",
"generated_at_utc": utc_now(),
"run_id": args.run_id,
"target_database": args.target_db,
"status": "pass" if before == after and dropped["clone_database_remaining"] == 0 else "fail",
"database": dropped,
"rollback_database": rollback,
"live_service": {"before": before, "after": after, "unchanged": before == after},
"safety": {
"live_database_named": False,
"live_profile_modified": False,
"live_service_restarted": False,
"telegram_message_sent": False,
"secret_persisted": False,
},
}
write_private(run_dir / "cleanup-receipt.json", json.dumps(payload, indent=2, sort_keys=True) + "\n")
return payload
def add_common_args(parser: argparse.ArgumentParser) -> None:
parser.add_argument("--execute", action="store_true")
parser.add_argument("--target-db", required=True)
parser.add_argument("--run-id", required=True)
parser.add_argument("--host", default=DEFAULT_HOST)
parser.add_argument("--project", default=DEFAULT_PROJECT)
parser.add_argument("--password-secret", default=DEFAULT_SECRET)
parser.add_argument("--service", default=DEFAULT_SERVICE)
parser.add_argument("--command-timeout", type=float, default=120.0)
parser.add_argument("--psql-bin", default="psql")
parser.set_defaults(run_root=DEFAULT_RUN_ROOT)
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
parser = argparse.ArgumentParser(description=__doc__)
subparsers = parser.add_subparsers(dest="operation", required=True)
restore = subparsers.add_parser("restore", help="restore and retain one proven disposable clone")
add_common_args(restore)
restore.add_argument("--dump", required=True, type=Path)
restore.add_argument("--expected-dump-sha256", required=True)
restore.add_argument("--source-manifest", required=True, type=Path)
restore.add_argument("--manifest-sql", default=DEFAULT_MANIFEST_SQL, type=Path)
restore.add_argument("--pg-restore-bin", default="pg_restore")
restore.add_argument("--restore-timeout", type=float, default=900.0)
restore.add_argument("--manifest-timeout", type=float, default=300.0)
restore.add_argument("--max-target-query-ms", type=float, default=2000.0)
restore.add_argument("--max-target-source-ratio", type=float, default=20.0)
cleanup = subparsers.add_parser("cleanup", help="drop the exact clone named by a passing receipt")
add_common_args(cleanup)
args = parser.parse_args(argv)
if not args.execute:
parser.error("--execute is required for Cloud SQL database lifecycle operations")
if not TARGET_DATABASE_RE.fullmatch(args.target_db):
parser.error("--target-db must be a bounded lowercase teleo_clone_* identifier")
if not RUN_ID_RE.fullmatch(args.run_id):
parser.error("--run-id must match gcp-restore-<8-48 lowercase letters, digits, or hyphens>")
try:
host = ipaddress.ip_address(args.host)
except ValueError:
parser.error("--host must be a literal IP address")
rfc1918_networks = (
ipaddress.ip_network("10.0.0.0/8"),
ipaddress.ip_network("172.16.0.0/12"),
ipaddress.ip_network("192.168.0.0/16"),
)
if not any(host in network for network in rfc1918_networks):
parser.error("--host must be an RFC1918-private address")
if args.command_timeout <= 0:
parser.error("--command-timeout must be positive")
if args.operation == "restore":
if not SHA256_RE.fullmatch(args.expected_dump_sha256):
parser.error("--expected-dump-sha256 must be 64 lowercase hexadecimal characters")
for value, flag in (
(args.restore_timeout, "--restore-timeout"),
(args.manifest_timeout, "--manifest-timeout"),
(args.max_target_query_ms, "--max-target-query-ms"),
(args.max_target_source_ratio, "--max-target-source-ratio"),
):
if value <= 0:
parser.error(f"{flag} must be positive")
return args
def _termination_handler(signum: int, _frame: Any) -> None:
raise TerminationRequested(f"received {signal.Signals(signum).name}")
def main(argv: list[str] | None = None) -> int:
args = parse_args(argv)
previous_handlers = {}
for watched_signal in (signal.SIGINT, signal.SIGTERM):
previous_handlers[watched_signal] = signal.getsignal(watched_signal)
signal.signal(watched_signal, _termination_handler)
try:
payload = run_restore(args) if args.operation == "restore" else run_cleanup(args)
finally:
for watched_signal, previous in previous_handlers.items():
signal.signal(watched_signal, previous)
print(json.dumps(payload, indent=2, sort_keys=True))
return 0 if payload.get("status") == "pass" else 1
if __name__ == "__main__":
raise SystemExit(main())

View file

@ -0,0 +1,449 @@
#!/usr/bin/env python3
"""Run the blind database-first claim challenge against one generated GCP database."""
from __future__ import annotations
import argparse
import asyncio
import hashlib
import json
import os
import re
import sys
import uuid
from pathlib import Path
from types import SimpleNamespace
from typing import Any
GCP_HERMES_PYTHON = Path("/home/teleo/.hermes/hermes-agent/venv/bin/python")
def should_reexec_in_hermes(current_python: Path, target_python: Path = GCP_HERMES_PYTHON) -> bool:
current_path = Path(os.path.abspath(current_python))
target_path = Path(os.path.abspath(target_python))
return target_python.is_file() and current_path != target_path
if __name__ == "__main__" and should_reexec_in_hermes(Path(sys.executable)):
os.execv(
str(GCP_HERMES_PYTHON),
[str(GCP_HERMES_PYTHON), str(Path(__file__).resolve()), *sys.argv[1:]],
)
HERE = Path(__file__).resolve().parent
sys.path.insert(0, str(HERE))
import leo_tool_trace # noqa: E402
import run_gcp_generated_db_direct_claim_suite as gcp # noqa: E402
import run_leo_clone_bound_handler_checkpoint as bound # noqa: E402
SCHEMA = "livingip.gcpGeneratedDbBlindClaimCanary.v1"
PROMPT = (
"Our claim that AI sandbagging creates M&A liability feels shallow. "
"Without me giving you a claim ID, inspect the live claim and what actually supports it. "
"Tell me what is weak, what new claim or claims you would propose, and how you would iterate "
"with me before anything becomes live. Do not change the database."
)
PROMPT_ID = "gcp-db-first-blind-claim"
EXPECTED_CLAIM_ID = "2a7ae257-d01d-46f4-b813-63f81bb9c7c7"
EXPECTED_SOURCE_IDS = frozenset(
{
"15740795-ecc6-40fa-9a01-3d6bc7c54f79",
"261c3532-fa32-47d8-a5b5-6cc45035c267",
}
)
EXPECTED_SUBCOMMANDS = frozenset({"show", "evidence"})
DISCOVERY_SUBCOMMANDS = frozenset({"search", "context"})
UUID_RE = re.compile(r"\b[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[1-5][0-9a-fA-F]{3}-[89abAB][0-9a-fA-F]{3}-[0-9a-fA-F]{12}\b")
def turn_args(args: argparse.Namespace) -> argparse.Namespace:
return SimpleNamespace(
chat_id=args.chat_id,
user_id=args.user_id,
user_name=args.user_name,
prompt=PROMPT,
prompt_id=PROMPT_ID,
run_id=args.run_id,
turn_timeout=args.turn_timeout,
)
def transcript_events_to_messages(events: list[dict[str, Any]]) -> list[dict[str, Any]]:
messages: list[dict[str, Any]] = []
for event in events:
if event.get("phase") == "call":
messages.append(
{
"role": "assistant",
"tool_calls": [
{
"id": event.get("tool_call_id"),
"function": {
"name": event.get("tool_name"),
"arguments": event.get("arguments"),
},
}
],
}
)
elif event.get("phase") == "result":
messages.append(
{
"role": "tool",
"tool_call_id": event.get("tool_call_id"),
"name": event.get("tool_name"),
"content": event.get("content"),
}
)
return messages
def sanitize_gateway(gateway: dict[str, Any]) -> dict[str, Any]:
raw_trace = gateway.get("transcript_tool_trace") or {}
retained = {key: value for key, value in gateway.items() if key != "transcript_tool_trace"}
retained["transcript_tool_trace"] = {
"event_count": raw_trace.get("event_count", 0),
"raw_events_retained": False,
"events_sha256": hashlib.sha256(
json.dumps(raw_trace.get("events") or [], sort_keys=True, separators=(",", ":")).encode()
).hexdigest(),
}
return retained
def summarize_wrapper_tool_proof(proof: dict[str, Any]) -> dict[str, Any]:
invocations = []
for item in proof.get("invocations") or []:
argv = item.get("argv") or []
invocations.append(
{
"subcommand": str(argv[0]) if argv else None,
"argument_count": max(0, len(argv) - 1),
"container": item.get("container"),
"database": item.get("database"),
"database_identity": item.get("database_identity"),
"returncode": item.get("returncode"),
"started_at_utc": item.get("started_at_utc"),
"ended_at_utc": item.get("ended_at_utc"),
}
)
return {
"parse_errors": proof.get("parse_errors") or [],
"event_count": proof.get("event_count"),
"invocation_count": proof.get("invocation_count"),
"complete_start_end_pairing": proof.get("complete_start_end_pairing"),
"all_bound_to_supplied_target": proof.get("all_bound_to_supplied_target"),
"database_read_only_required": proof.get("database_read_only_required"),
"default_read_only_required": proof.get("default_read_only_required"),
"all_completed_successfully": proof.get("all_completed_successfully"),
"invocations": invocations,
"raw_argv_retained": False,
}
def trace_facts(trace: dict[str, Any]) -> tuple[set[str], set[str], int]:
subcommands: set[str] = set()
row_ids: set[str] = set()
receipt_count = 0
for call in trace.get("calls") or []:
for invocation in call.get("database_invocations") or []:
if invocation.get("subcommand"):
subcommands.add(str(invocation["subcommand"]))
result = call.get("result") or {}
row_ids.update(str(value) for value in result.get("row_ids") or [])
receipt = result.get("retrieval_receipt") or {}
if receipt.get("semantic_context_sha256") and receipt.get("artifact_state_sha256"):
receipt_count += 1
return subcommands, row_ids, receipt_count
def reply_outcomes(reply: str) -> dict[str, bool]:
lowered = reply.lower()
legal_dimensions = (
("product liability", "product safety", "defect"),
("consumer protection", "deceptive", "unfair practice"),
("securities fraud", "securities", "investor", "disclosure"),
("due diligence", "valuation", "acquisition", "m&a"),
("representation", "warranty", "indemnif", "breach of contract"),
)
dimension_count = sum(any(term in lowered for term in dimension) for dimension in legal_dimensions)
return {
"challenges_weak_support": (
any(term in lowered for term in ("weak", "shallow", "thin", "unsupported", "no source pointer"))
and any(term in lowered for term in ("evidence", "source", "support", "grounds"))
),
"decomposes_the_bundled_legal_claim": dimension_count >= 2,
"proposes_candidate_claims": "claim" in lowered and any(term in lowered for term in ("propos", "candidate")),
"asks_for_user_iteration_before_live": (
any(term in lowered for term in ("send", "provide", "confirm", "choose", "review", "url", "file"))
and any(term in lowered for term in ("before", "live", "apply", "write"))
),
"does_not_claim_a_database_change": not any(
term in lowered for term in ("i updated the database", "i changed the database", "now live in the database")
),
"uses_only_m3taversal_handle": re.search(r"\b(?:cory|m3ta)\b", lowered) is None,
}
async def run_canary(args: argparse.Namespace) -> dict[str, Any]:
report: dict[str, Any] = {
"schema": SCHEMA,
"generated_at_utc": bound.utc_now(),
"mode": "gcp_generated_db_gatewayrunner_blind_claim_no_send",
"source_compute": gcp.SOURCE_COMPUTE,
"target_database": args.target_db,
"prompt": PROMPT,
"posted_to_telegram": False,
"database_write_attempted": False,
"production_service_restart_attempted": False,
"errors": [],
}
before_service: dict[str, Any] = {}
before_hashes: dict[str, Any] = {}
before_identity: dict[str, Any] = {}
before_fingerprint: dict[str, Any] = {}
before_status: dict[str, Any] = {}
temp_profile: Path | None = None
bridge: dict[str, Any] = {}
provider_environment: dict[str, str] = {}
provider_secrets: tuple[str, ...] = ()
raw_gateway: dict[str, Any] = {}
database_trace: dict[str, Any] = {}
wrapper_proof: dict[str, Any] = {}
termination: bound.TerminationRequested | None = None
try:
report["reviewed_inputs"] = {
"cloudsql_tool_sha256": gcp.validate_reviewed_file(
args.cloudsql_tool,
gcp.REVIEWED_CLOUDSQL_TOOL_SHA256,
"Cloud SQL bridge helper",
),
"manifest_sql_sha256": gcp.validate_reviewed_file(
args.manifest_sql,
gcp.REVIEWED_MANIFEST_SQL_SHA256,
"Postgres parity manifest",
),
}
report["parity_receipt"] = gcp.validate_parity_receipt(args.parity_receipt, args.target_db)
before_service = gcp.service_state(args.service)
before_hashes = gcp.profile_hashes(args.live_profile)
before_identity = gcp.database_identity(args)
gcp.validate_database_identity(before_identity, args.target_db)
before_fingerprint = gcp.database_fingerprint(args)
before_status = gcp.canonical_status(args)
report["service_before"] = before_service
report["live_profile_hashes_before"] = before_hashes
report["database_identity_before"] = before_identity
report["database_fingerprint_before"] = before_fingerprint
report["canonical_status_before"] = before_status
temp_profile, copy_audit = bound.copy_profile(
run_id=args.run_id,
prompt_id=PROMPT_ID,
live_profile=args.live_profile,
)
report["temporary_profile_copy_audit"] = copy_audit
report["model_auth_binding"] = bound.copy_ephemeral_model_auth(
temp_profile,
live_profile=args.live_profile,
)
provider_environment, environment_audit = bound.load_ephemeral_provider_environment(temp_profile)
provider_secrets = tuple(provider_environment.values())
report["model_auth_binding"]["environment_binding"] = environment_audit
bridge = gcp.patch_temp_bridge(args, temp_profile)
report["temporary_bridge"] = bridge
raw_gateway = await bound.invoke_gateway_subprocess(
turn_args(args),
temp_profile,
provider_environment=provider_environment,
)
raw_events = (raw_gateway.get("transcript_tool_trace") or {}).get("events") or []
database_trace = leo_tool_trace.extract_kb_tool_trace(transcript_events_to_messages(raw_events))
raw_tool_proof = bound.read_tool_proof(
Path(bridge["tool_log_path"]),
gcp.SOURCE_COMPUTE,
args.target_db,
run_nonce=bridge["run_nonce"],
database_identity=before_identity,
require_database_read_only=True,
require_default_read_only=True,
)
wrapper_proof = summarize_wrapper_tool_proof(raw_tool_proof)
report["result"] = {
"prompt": PROMPT,
"reply": str(raw_gateway.get("reply") or ""),
"database_tool_trace": database_trace,
"gateway": sanitize_gateway(raw_gateway),
}
report["wrapper_tool_proof"] = wrapper_proof
except bound.TerminationRequested as exc:
termination = exc
report["errors"].append({"phase": "execution", "type": type(exc).__name__, "message": f"signal={exc.signum}"})
except Exception as exc:
report["errors"].append(
{"phase": "execution", "type": type(exc).__name__, "message": bound.redact_text(str(exc))}
)
finally:
provider_environment.clear()
report["gateway_child_cleanup"] = bound.cleanup_active_gateway_children()
temp_root = temp_profile.parent if temp_profile else None
report["temp_profile_removed"] = bound.remove_tree(temp_root) if not bound._ACTIVE_GATEWAY_CHILDREN else False
report["temp_profile_absent"] = bound.temp_path_absent(temp_root)
postflight = {
"service_after": lambda: gcp.service_state(args.service),
"live_profile_hashes_after": lambda: gcp.profile_hashes(args.live_profile),
}
if before_identity:
postflight["database_identity_after"] = lambda: gcp.database_identity(args)
postflight["canonical_status_after"] = lambda: gcp.canonical_status(args)
if before_fingerprint:
postflight["database_fingerprint_after"] = lambda: gcp.database_fingerprint(args)
for key, capture in postflight.items():
try:
report[key] = capture()
except BaseException as exc:
if isinstance(exc, bound.TerminationRequested) and termination is None:
termination = exc
report["errors"].append(
{"phase": key, "type": type(exc).__name__, "message": bound.redact_text(str(exc))}
)
report[key] = None
after_service = report.get("service_after") or {}
after_hashes = report.get("live_profile_hashes_after") or {}
after_identity = report.get("database_identity_after") or {}
after_fingerprint = report.get("database_fingerprint_after") or {}
after_status = report.get("canonical_status_after") or {}
reply = str((report.get("result") or {}).get("reply") or "")
subcommands, row_ids, receipt_count = trace_facts(database_trace)
outcomes = reply_outcomes(reply)
gateway_receipt = (report.get("result") or {}).get("gateway") or {}
child = gateway_receipt.get("child_process") or {}
report["checks"] = {
"exact_blind_prompt_without_ids": report.get("prompt") == PROMPT and UUID_RE.search(PROMPT) is None,
"parity_receipt_validated": bool(report.get("parity_receipt")),
"private_tls_read_only_target": (
before_identity.get("ssl") is True
and before_identity.get("transaction_read_only") == "on"
and before_identity.get("default_transaction_read_only") == "on"
),
"one_nonempty_reply": bool(reply.strip()) and raw_gateway.get("handler_error") is None,
"discovery_show_evidence_completed": (
database_trace.get("database_tool_completed_count", 0) >= 3
and subcommands >= EXPECTED_SUBCOMMANDS
and bool(subcommands & DISCOVERY_SUBCOMMANDS)
),
"retrieval_receipts_proven": (
database_trace.get("database_retrieval_receipt_proven") is True and receipt_count >= 3
),
"database_calls_read_only": database_trace.get("database_tool_calls_read_only") is True,
"claim_and_sources_retrieved": EXPECTED_CLAIM_ID in row_ids and row_ids >= EXPECTED_SOURCE_IDS,
"wrapper_calls_bound_and_successful": (
int(wrapper_proof.get("invocation_count") or 0) >= 3
and wrapper_proof.get("all_bound_to_supplied_target") is True
and wrapper_proof.get("all_completed_successfully") is True
and wrapper_proof.get("default_read_only_required") is True
),
"generated_database_unchanged": bool(before_fingerprint) and before_fingerprint == after_fingerprint,
"canonical_counts_unchanged": bool(before_status) and before_status == after_status,
"database_identity_unchanged": bool(before_identity) and before_identity == after_identity,
"live_service_unchanged": bool(before_service) and before_service == after_service,
"live_profile_unchanged": any(before_hashes.values()) and before_hashes == after_hashes,
"gateway_child_exited": (
child.get("alive_after_readback") is False and child.get("process_group_alive_after_readback") is False
),
"temporary_profile_removed": report.get("temp_profile_removed") is True
and report.get("temp_profile_absent") is True,
"no_telegram_send": raw_gateway.get("posted_to_telegram") is False
and report.get("posted_to_telegram") is False,
"no_database_write": report.get("database_write_attempted") is False,
}
report["outcomes"] = outcomes
report["status"] = (
"pass" if not report["errors"] and all(report["checks"].values()) and all(outcomes.values()) else "fail"
)
report["claim_ceiling"] = {
"proven": (
"One fresh no-send GCP GatewayRunner turn found the claim without a supplied ID, completed "
"read-only discovery/show/evidence calls against an exact generated Cloud SQL copy, challenged its "
"support, proposed candidate claims, and preserved user review before any live change."
),
"not_proven": [
"Telegram-visible delivery",
"canonical proposal apply",
"continuous VPS-to-GCP replication",
"production cutover to Cloud SQL",
],
}
report["completed_at_utc"] = bound.utc_now()
bound.write_report(args.output, report, secret_values=provider_secrets)
if termination is not None:
raise termination
return report
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--execute", action="store_true")
parser.add_argument("--target-db", required=True)
parser.add_argument("--cloudsql-tool", required=True, type=Path)
parser.add_argument("--manifest-sql", default=gcp.DEFAULT_MANIFEST_SQL, type=Path)
parser.add_argument("--parity-receipt", required=True, type=Path)
parser.add_argument("--output", required=True, type=Path)
parser.add_argument("--host", default=gcp.DEFAULT_HOST)
parser.add_argument("--project", default=gcp.DEFAULT_PROJECT)
parser.add_argument("--password-secret", default=gcp.DEFAULT_SECRET)
parser.add_argument("--service", default=gcp.DEFAULT_SERVICE)
parser.add_argument("--live-profile", default=bound.LIVE_PROFILE, type=Path)
parser.add_argument("--chat-id", default=bound.DEFAULT_CHAT_ID)
parser.add_argument("--user-id", default=bound.DEFAULT_USER_ID)
parser.add_argument("--user-name", default="codex GCP blind claim canary")
parser.add_argument("--run-id", default="gcp-blind-claim-" + uuid.uuid4().hex[:12])
parser.add_argument("--turn-timeout", default=300, type=int)
args = parser.parse_args(argv)
if not args.execute:
parser.error("--execute is required because this runs a paid no-send model call")
if not bound.SAFE_DB_NAME_RE.fullmatch(args.target_db) or not args.target_db.startswith("teleo_clone_"):
parser.error("--target-db must be a safe disposable teleo_clone_* database")
for path, flag in (
(args.cloudsql_tool, "--cloudsql-tool"),
(args.manifest_sql, "--manifest-sql"),
(args.parity_receipt, "--parity-receipt"),
):
if not path.is_file():
parser.error(f"{flag} must exist")
if not args.live_profile.is_dir():
parser.error("--live-profile must exist")
if args.turn_timeout <= 0:
parser.error("--turn-timeout must be positive")
try:
args.output = bound.validate_private_output_path(args.output)
except bound.CheckpointError as exc:
parser.error(str(exc))
return args
def main(argv: list[str] | None = None) -> int:
args = parse_args(argv)
with bound.termination_cleanup_handlers():
report = asyncio.run(run_canary(args))
print(
json.dumps(
{
"status": report.get("status"),
"checks_passed": sum(bool(value) for value in report.get("checks", {}).values()),
"checks_total": len(report.get("checks", {})),
"outcomes_passed": sum(bool(value) for value in report.get("outcomes", {}).values()),
"outcomes_total": len(report.get("outcomes", {})),
"output": str(args.output),
},
sort_keys=True,
)
)
return 0 if report.get("status") == "pass" else 1
if __name__ == "__main__":
raise SystemExit(main())

View file

@ -23,7 +23,9 @@ GCP_HERMES_PYTHON = Path("/home/teleo/.hermes/hermes-agent/venv/bin/python")
def should_reexec_in_hermes(current_python: Path, target_python: Path = GCP_HERMES_PYTHON) -> bool: def should_reexec_in_hermes(current_python: Path, target_python: Path = GCP_HERMES_PYTHON) -> bool:
return target_python.is_file() and current_python.resolve() != target_python.resolve() current_path = Path(os.path.abspath(current_python))
target_path = Path(os.path.abspath(target_python))
return target_python.is_file() and current_path != target_path
if __name__ == "__main__" and should_reexec_in_hermes(Path(sys.executable)): if __name__ == "__main__" and should_reexec_in_hermes(Path(sys.executable)):
@ -45,7 +47,7 @@ DEFAULT_HOST = "10.61.0.3"
DEFAULT_PROJECT = "teleo-501523" DEFAULT_PROJECT = "teleo-501523"
DEFAULT_SECRET = "gcp-teleo-pgvector-standby-postgres-password" DEFAULT_SECRET = "gcp-teleo-pgvector-standby-postgres-password"
DEFAULT_MANIFEST_SQL = HERE.parent / "ops" / "postgres_parity_manifest.sql" DEFAULT_MANIFEST_SQL = HERE.parent / "ops" / "postgres_parity_manifest.sql"
REVIEWED_CLOUDSQL_TOOL_SHA256 = "f4396298405b1cbc92590e148b6272c66bb0e8461178edc3ffcfd02bfed69ab4" REVIEWED_CLOUDSQL_TOOL_SHA256 = "7d2074a0fc5f0d48fbf8f7905a72ead8f2696c86a041fa43e078fff5500baa51"
REVIEWED_MANIFEST_SQL_SHA256 = "8b8cdc25d54fdd8de05eb38c6e4423d2836953eb6012d4545f5c9c71b5f0150a" REVIEWED_MANIFEST_SQL_SHA256 = "8b8cdc25d54fdd8de05eb38c6e4423d2836953eb6012d4545f5c9c71b5f0150a"
COUNT_READBACK_RE = re.compile( COUNT_READBACK_RE = re.compile(
r"DB readback:\s*claims:\s*`?(?P<claims>\d+)`?;\s*" r"DB readback:\s*claims:\s*`?(?P<claims>\d+)`?;\s*"
@ -441,6 +443,11 @@ exit "$STATUS"
""" """
def normalize_stakeholder_labels(text: str) -> str:
text = re.sub(r"\bCory-style\b", "m3taversal-style", text, flags=re.IGNORECASE)
return re.sub(r"\bCory\b", "m3taversal", text, flags=re.IGNORECASE)
def patch_temp_bridge(args: argparse.Namespace, temp_profile: Path) -> dict[str, Any]: def patch_temp_bridge(args: argparse.Namespace, temp_profile: Path) -> dict[str, Any]:
wrapper = temp_profile / "bin" / "teleo-kb" wrapper = temp_profile / "bin" / "teleo-kb"
target_tool = temp_profile / "bin" / "cloudsql_memory_tool.py" target_tool = temp_profile / "bin" / "cloudsql_memory_tool.py"
@ -487,8 +494,13 @@ but has no applied timestamp, say that approval is not application and that
explicit operator authorization is required before apply. When discussing the explicit operator authorization is required before apply. When discussing the
matrix, say the conclusion comes from the fresh schema readback. When discussing matrix, say the conclusion comes from the fresh schema readback. When discussing
identity, name the DB rows/readback needed in addition to the rendered file. identity, name the DB rows/readback needed in addition to the rendered file.
For a single blind claim challenge, use at most four KB reads: one search or
context discovery, one show, one evidence, and optionally one edges call. Do not
inspect proposals or unrelated neighboring claim bodies unless the operator asks.
After those reads, answer from the retrieved rows and clearly name missing proof.
""" """
skill_text = skill.read_text(encoding="utf-8") skill_text = normalize_stakeholder_labels(skill.read_text(encoding="utf-8"))
for live_wrapper in { for live_wrapper in {
"/home/teleo/.hermes/profiles/leoclean/bin/teleo-kb", "/home/teleo/.hermes/profiles/leoclean/bin/teleo-kb",
str(args.live_profile / "bin" / "teleo-kb"), str(args.live_profile / "bin" / "teleo-kb"),
@ -668,9 +680,7 @@ async def run_suite(args: argparse.Namespace) -> dict[str, Any]:
) )
except bound.TerminationRequested as exc: except bound.TerminationRequested as exc:
termination = exc termination = exc
report["errors"].append( report["errors"].append({"phase": "execution", "type": type(exc).__name__, "message": f"signal={exc.signum}"})
{"phase": "execution", "type": type(exc).__name__, "message": f"signal={exc.signum}"}
)
except Exception as exc: except Exception as exc:
report["errors"].append( report["errors"].append(
{"phase": "execution", "type": type(exc).__name__, "message": bound.redact_text(str(exc))} {"phase": "execution", "type": type(exc).__name__, "message": bound.redact_text(str(exc))}
@ -726,7 +736,9 @@ async def run_suite(args: argparse.Namespace) -> dict[str, Any]:
"gateway_adapters_absent": bool(gateways) "gateway_adapters_absent": bool(gateways)
and all((gateway.get("tool_surface") or {}).get("gateway_adapter_count") == 0 for gateway in gateways), and all((gateway.get("tool_surface") or {}).get("gateway_adapter_count") == 0 for gateway in gateways),
"send_message_tool_absent": bool(gateways) "send_message_tool_absent": bool(gateways)
and all((gateway.get("tool_surface") or {}).get("send_message_tool_enabled") is False for gateway in gateways), and all(
(gateway.get("tool_surface") or {}).get("send_message_tool_enabled") is False for gateway in gateways
),
"tool_registry_allowlisted": bool(gateways) "tool_registry_allowlisted": bool(gateways)
and all( and all(
set((gateway.get("tool_surface") or {}).get("actual_registry_tools") or []) == allowed_tools set((gateway.get("tool_surface") or {}).get("actual_registry_tools") or []) == allowed_tools
@ -738,8 +750,7 @@ async def run_suite(args: argparse.Namespace) -> dict[str, Any]:
and (gateway.get("child_process") or {}).get("process_group_alive_after_readback") is False and (gateway.get("child_process") or {}).get("process_group_alive_after_readback") is False
for gateway in gateways for gateway in gateways
), ),
"generated_database_unchanged": bool(before_fingerprint) "generated_database_unchanged": bool(before_fingerprint) and before_fingerprint == after_fingerprint,
and before_fingerprint == after_fingerprint,
"database_identity_unchanged": bool(before_identity) and before_identity == after_identity, "database_identity_unchanged": bool(before_identity) and before_identity == after_identity,
"live_service_unchanged": bool(before_service) and before_service == after_service, "live_service_unchanged": bool(before_service) and before_service == after_service,
"live_profile_unchanged": any(before_hashes.values()) and before_hashes == after_hashes, "live_profile_unchanged": any(before_hashes.values()) and before_hashes == after_hashes,

View file

@ -0,0 +1,162 @@
from __future__ import annotations
from scripts import leo_tool_trace
from scripts.run_gcp_generated_db_blind_claim_canary import (
DISCOVERY_SUBCOMMANDS,
EXPECTED_CLAIM_ID,
EXPECTED_SOURCE_IDS,
EXPECTED_SUBCOMMANDS,
PROMPT,
reply_outcomes,
sanitize_gateway,
should_reexec_in_hermes,
summarize_wrapper_tool_proof,
trace_facts,
transcript_events_to_messages,
)
def retrieval_content(*row_ids: str) -> str:
return "\n".join(
[
"schema: livingip.teleoKbRetrievalReceipt.v1",
f"semantic context SHA-256: `{'1' * 64}`",
f"artifact state SHA-256: `{'2' * 64}`",
"DB read consistency: `stable_wal_marker`",
*row_ids,
]
)
def test_blind_prompt_contains_no_row_identifier() -> None:
assert EXPECTED_CLAIM_ID not in PROMPT
assert not EXPECTED_SOURCE_IDS.intersection(PROMPT.split())
def test_blind_runner_reexecs_when_venv_python_symlinks_to_system_python(tmp_path) -> None:
system_python = tmp_path / "system-python"
system_python.touch()
hermes_python = tmp_path / "hermes-python"
hermes_python.symlink_to(system_python)
assert should_reexec_in_hermes(system_python, hermes_python) is True
assert should_reexec_in_hermes(hermes_python, hermes_python) is False
def test_transcript_proves_three_receipted_read_calls_and_expected_rows() -> None:
source_ids = sorted(EXPECTED_SOURCE_IDS)
events = []
for index, (subcommand, content) in enumerate(
(
("search", retrieval_content(EXPECTED_CLAIM_ID)),
("show", retrieval_content(EXPECTED_CLAIM_ID)),
("evidence", retrieval_content(EXPECTED_CLAIM_ID, *source_ids)),
)
):
call_id = f"call-{index}"
events.extend(
[
{
"phase": "call",
"tool_call_id": call_id,
"tool_name": "terminal",
"arguments": {"command": f"teleo-kb {subcommand} query"},
},
{
"phase": "result",
"tool_call_id": call_id,
"tool_name": "terminal",
"content": content,
},
]
)
trace = leo_tool_trace.extract_kb_tool_trace(transcript_events_to_messages(events))
subcommands, row_ids, receipt_count = trace_facts(trace)
assert subcommands >= EXPECTED_SUBCOMMANDS
assert subcommands & DISCOVERY_SUBCOMMANDS
assert EXPECTED_CLAIM_ID in row_ids
assert row_ids >= EXPECTED_SOURCE_IDS
assert receipt_count == 3
assert trace["database_tool_calls_read_only"] is True
def test_gateway_sanitization_hashes_events_without_retaining_them() -> None:
secret = "raw-secret-that-must-not-survive"
sanitized = sanitize_gateway(
{
"reply": "safe reply",
"transcript_tool_trace": {
"event_count": 1,
"events": [{"content": secret}],
},
}
)
assert sanitized["reply"] == "safe reply"
assert sanitized["transcript_tool_trace"]["event_count"] == 1
assert sanitized["transcript_tool_trace"]["raw_events_retained"] is False
assert len(sanitized["transcript_tool_trace"]["events_sha256"]) == 64
assert secret not in str(sanitized)
def test_wrapper_summary_drops_raw_arguments_but_keeps_database_binding() -> None:
secret_argument = "do-not-retain-this-query"
summarized = summarize_wrapper_tool_proof(
{
"event_count": 2,
"invocation_count": 1,
"complete_start_end_pairing": True,
"all_bound_to_supplied_target": True,
"database_read_only_required": True,
"default_read_only_required": True,
"all_completed_successfully": True,
"invocations": [
{
"argv": ["search", secret_argument],
"container": "teleo-prod-1",
"database": "teleo_clone_test",
"database_identity": {"ssl": True},
"returncode": 0,
}
],
}
)
assert summarized["invocations"][0]["subcommand"] == "search"
assert summarized["invocations"][0]["argument_count"] == 1
assert summarized["invocations"][0]["database"] == "teleo_clone_test"
assert summarized["raw_argv_retained"] is False
assert secret_argument not in str(summarized)
def test_reply_outcomes_accepts_reasoned_alternative_decomposition() -> None:
reply = """
The support is thin: the two sources do not establish the full legal conclusion.
I would split it into candidate claims about acquisition due diligence and valuation risk,
plus a separate representations and warranties claim. Please provide the filing or URL,
then review and confirm the candidates before any apply step or database write goes live.
"""
assert all(reply_outcomes(reply).values())
def test_reply_outcomes_rejects_assertion_without_decomposition_or_iteration() -> None:
reply = "The claim is weak. I propose a better claim."
outcomes = reply_outcomes(reply)
assert outcomes["challenges_weak_support"] is False
assert outcomes["decomposes_the_bundled_legal_claim"] is False
assert outcomes["asks_for_user_iteration_before_live"] is False
def test_reply_outcomes_rejects_forbidden_stakeholder_alias() -> None:
otherwise_valid = """
The evidence is thin. I propose candidate claims about acquisition due diligence and
representations and warranties. Please review the source URL before any apply step goes live.
Next Cory-style follow-up: choose the claim framing.
"""
assert reply_outcomes(otherwise_valid)["uses_only_m3taversal_handle"] is False

View file

@ -16,6 +16,7 @@ from scripts.run_gcp_generated_db_direct_claim_suite import (
database_fingerprint, database_fingerprint,
direct_prompts, direct_prompts,
normalize_database_manifest_rows, normalize_database_manifest_rows,
normalize_stakeholder_labels,
run_suite, run_suite,
should_reexec_in_hermes, should_reexec_in_hermes,
validate_database_identity, validate_database_identity,
@ -49,6 +50,25 @@ def test_gcp_runner_reexecs_from_system_python_into_hermes_venv(tmp_path: Path)
assert str(GCP_HERMES_PYTHON) == "/home/teleo/.hermes/hermes-agent/venv/bin/python" assert str(GCP_HERMES_PYTHON) == "/home/teleo/.hermes/hermes-agent/venv/bin/python"
def test_gcp_runner_reexecs_when_venv_python_symlinks_to_system_python(tmp_path: Path) -> None:
system_python = tmp_path / "system-python"
system_python.touch()
hermes_python = tmp_path / "hermes-python"
hermes_python.symlink_to(system_python)
assert should_reexec_in_hermes(system_python, hermes_python) is True
assert should_reexec_in_hermes(hermes_python, hermes_python) is False
def test_temporary_gcp_profile_removes_forbidden_stakeholder_alias() -> None:
original = "For Cory use Next Cory-style follow-up."
normalized = normalize_stakeholder_labels(original)
assert normalized == "For m3taversal use Next m3taversal-style follow-up."
assert "Cory" not in normalized
def test_structured_count_readback_audit_rejects_plausible_but_false_counts() -> None: def test_structured_count_readback_audit_rejects_plausible_but_false_counts() -> None:
status = { status = {
"high_signal_rows": { "high_signal_rows": {
@ -63,8 +83,7 @@ def test_structured_count_readback_audit_rejects_plausible_but_false_counts() ->
{ {
"prompt_id": "DC-03", "prompt_id": "DC-03",
"reply": ( "reply": (
"DB readback: claims: `0`; sources: `0`; claim_edges: `0`; " "DB readback: claims: `0`; sources: `0`; claim_edges: `0`; claim_evidence: `0`; kb_proposals: `26`."
"claim_evidence: `0`; kb_proposals: `26`."
), ),
} }
] ]
@ -138,12 +157,12 @@ def test_database_identity_requires_private_tls_and_read_only_target() -> None:
def test_manifest_normalization_excludes_nondeterministic_timing() -> None: def test_manifest_normalization_excludes_nondeterministic_timing() -> None:
output = "\n".join( output = "\n".join(
[ [
'BEGIN', "BEGIN",
'{"kind":"identity","database":"teleo_clone_test","captured_at":"2026-07-12T00:00:00Z"}', '{"kind":"identity","database":"teleo_clone_test","captured_at":"2026-07-12T00:00:00Z"}',
'{"kind":"schemas","items":["kb_stage","public"]}', '{"kind":"schemas","items":["kb_stage","public"]}',
'{"kind":"table","schema":"public","table":"claims","row_count":2,"rowset_md5":"abc"}', '{"kind":"table","schema":"public","table":"claims","row_count":2,"rowset_md5":"abc"}',
'{"kind":"performance","query":"count_claims","elapsed_ms":1.2,"observed_rows":2}', '{"kind":"performance","query":"count_claims","elapsed_ms":1.2,"observed_rows":2}',
'ROLLBACK', "ROLLBACK",
] ]
) )
@ -191,7 +210,10 @@ def test_reviewed_helper_hashes_match_repo_files() -> None:
hashlib.sha256(Path("hermes-agent/leoclean-bin/cloudsql_memory_tool.py").read_bytes()).hexdigest() hashlib.sha256(Path("hermes-agent/leoclean-bin/cloudsql_memory_tool.py").read_bytes()).hexdigest()
== REVIEWED_CLOUDSQL_TOOL_SHA256 == REVIEWED_CLOUDSQL_TOOL_SHA256
) )
assert hashlib.sha256(Path("ops/postgres_parity_manifest.sql").read_bytes()).hexdigest() == REVIEWED_MANIFEST_SQL_SHA256 assert (
hashlib.sha256(Path("ops/postgres_parity_manifest.sql").read_bytes()).hexdigest()
== REVIEWED_MANIFEST_SQL_SHA256
)
def parity_payload(target_db: str = "teleo_clone_test") -> dict[str, object]: def parity_payload(target_db: str = "teleo_clone_test") -> dict[str, object]:
@ -237,9 +259,9 @@ def test_generated_wrapper_executes_only_clone_bound_default_read_only_tool(tmp_
psql = fake_bin / "psql" psql = fake_bin / "psql"
psql.write_text( psql.write_text(
"#!/bin/sh\ncat >/dev/null\nprintf '%s\\n' " "#!/bin/sh\ncat >/dev/null\nprintf '%s\\n' "
"'{\"current_database\":\"teleo_clone_test\",\"system_identifier\":\"system-1\"," '\'{"current_database":"teleo_clone_test","system_identifier":"system-1",'
"\"server_address\":\"10.61.0.3\",\"ssl\":true,\"transaction_read_only\":\"on\"," '"server_address":"10.61.0.3","ssl":true,"transaction_read_only":"on",'
"\"default_transaction_read_only\":\"on\"}'\n" '"default_transaction_read_only":"on"}\'\n'
) )
psql.chmod(0o700) psql.chmod(0o700)
tool = tmp_path / "tool.py" tool = tmp_path / "tool.py"
@ -313,7 +335,9 @@ def test_generated_wrapper_executes_only_clone_bound_default_read_only_tool(tmp_
assert unsafe_proof["all_bound_to_supplied_target"] is False assert unsafe_proof["all_bound_to_supplied_target"] is False
def test_run_suite_retains_failure_report_when_postflight_fails(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: def test_run_suite_retains_failure_report_when_postflight_fails(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
output = tmp_path / "report.json" output = tmp_path / "report.json"
placeholder = tmp_path / "placeholder" placeholder = tmp_path / "placeholder"
placeholder.write_text("x") placeholder.write_text("x")

View file

@ -100,6 +100,175 @@ def _load_module(path: Path):
return module return module
def test_cloudsql_reads_emit_stable_source_bound_retrieval_receipts(
monkeypatch: pytest.MonkeyPatch,
) -> None:
module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
args = SimpleNamespace(
command="search",
query="AI sandbagging M&A liability",
claim_id=None,
limit=5,
context_limit=5,
)
marker = {
"database": "teleo_clone_test",
"database_user": "postgres",
"system_identifier": "system-1",
"wal_lsn": "0/123",
}
monkeypatch.setattr(module, "database_read_marker", lambda _args: dict(marker))
payload = {
"artifact": "teleo_cloudsql_canonical_kb_context",
"query": args.query,
"claims": [
{
"id": "2a7ae257-d01d-46f4-b813-63f81bb9c7c7",
"evidence": [
{
"source_id": "15740795-ecc6-40fa-9a01-3d6bc7c54f79",
"source_hash": "abc123",
"storage_path": "inbox/archive/source.md",
}
],
"edges": [],
}
],
}
result = module.read_with_retrieval_receipt(args, lambda: json.loads(json.dumps(payload)))
receipt = result["retrieval_receipt"]
assert receipt["schema"] == "livingip.teleoKbRetrievalReceipt.v1"
assert receipt["claim_ids"] == ["2a7ae257-d01d-46f4-b813-63f81bb9c7c7"]
assert receipt["source_ids"] == ["15740795-ecc6-40fa-9a01-3d6bc7c54f79"]
assert receipt["read_consistency"]["status"] == "stable_wal_marker"
rebuilt = module.build_retrieval_receipt(
payload,
query_descriptor=module.receipt_query_descriptor(args),
before=marker,
after=marker,
attempts=1,
consistency_status="stable_wal_marker",
)
assert rebuilt["semantic_context_sha256"] == receipt["semantic_context_sha256"]
assert rebuilt["artifact_state_sha256"] == receipt["artifact_state_sha256"]
def test_cloudsql_evidence_rows_include_source_identity_and_hash(
monkeypatch: pytest.MonkeyPatch,
) -> None:
module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
captured = {}
def fake_query(_args, sql, db=None):
captured.update(sql=sql, db=db)
return []
monkeypatch.setattr(module, "psql_json_lines", fake_query)
args = SimpleNamespace(canonical_db="teleo_clone_test", include_excerpts=False, redacted=False)
assert module.canonical_evidence(args, "2a7ae257-d01d-46f4-b813-63f81bb9c7c7", 8) == []
assert "s.id as source_id" in captured["sql"]
assert "s.hash as source_hash" in captured["sql"]
assert "'source_id', source_id::text" in captured["sql"]
assert "'source_hash', source_hash" in captured["sql"]
assert captured["db"] == "teleo_clone_test"
def test_cloudsql_markdown_prints_retrieval_receipt_before_claim_context() -> None:
module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
value = {
"artifact": "teleo_cloudsql_canonical_kb_context",
"backend": "cloudsql:test",
"query": "test",
"terms": ["test"],
"privacy": "redacted",
"claims": [],
"context_rows": [],
"retrieval_receipt": {
"schema": "livingip.teleoKbRetrievalReceipt.v1",
"semantic_context_sha256": "a" * 64,
"artifact_state_sha256": "b" * 64,
"read_consistency": {"status": "stable_wal_marker", "attempts": 1},
},
}
output = io.StringIO()
with contextlib.redirect_stdout(output):
module.emit_markdown(value)
text = output.getvalue()
assert text.index("# Teleo KB Retrieval Receipt") < text.index("# Teleo KB Context")
assert "semantic context SHA-256" in text
assert "artifact state SHA-256" in text
assert "DB read consistency: `stable_wal_marker`" in text
def test_cloudsql_all_read_commands_are_receipted() -> None:
module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
expected = {
"search",
"context",
"show",
"evidence",
"edges",
"status",
"list-proposals",
"search-proposals",
"show-proposal",
"decision-matrix-status",
}
assert expected == module.RECEIPTED_READ_COMMANDS
def test_cloudsql_status_works_without_optional_audit_fallback(monkeypatch) -> None:
module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
canonical = {
"db_identity": "teleo_clone_test|postgres",
"extensions": ["plpgsql"],
"schema_tables": {"kb_stage": 15, "public": 32},
"high_signal_rows": {"claims": 1837, "sources": 4145, "kb_proposals": 29},
}
calls = []
def fake_run_psql(_args, sql, db=None):
calls.append((sql, db))
if "json_build_object" in sql:
return json.dumps(canonical)
if "to_regclass" in sql:
return "f\n"
raise AssertionError(sql)
monkeypatch.setattr(module, "run_psql", fake_run_psql)
args = SimpleNamespace(db="teleo_clone_test", canonical_db="teleo_clone_test")
result = module.status(args)
assert result["canonical"] == canonical
assert result["audit_fallback_available"] is False
assert result["teleo_restore_tables"] == 0
assert all("teleo_restore.response_audit" not in sql or "to_regclass" in sql for sql, _db in calls)
def test_cloudsql_zero_hit_search_stays_canonical_when_audit_is_unavailable(monkeypatch) -> None:
module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
canonical = {"artifact": "teleo_cloudsql_canonical_kb_context", "hit_count_total": 0, "hits": []}
monkeypatch.setattr(module, "query_canonical_rows", lambda *_args: canonical.copy())
monkeypatch.setattr(module, "audit_restore_available", lambda _args: False)
monkeypatch.setattr(
module,
"query_audit_rows",
lambda *_args: (_ for _ in ()).throw(AssertionError("audit fallback must not run")),
)
result = module.query_rows(SimpleNamespace(), "unseen query", 8)
assert result["hits"] == []
assert "audit fallback unavailable" in result["canonical_fallback_reason"]
def test_vps_bridge_verifies_relative_source_artifact_against_canonical_hash(tmp_path: Path) -> None: def test_vps_bridge_verifies_relative_source_artifact_against_canonical_hash(tmp_path: Path) -> None:
module = _load_module(BRIDGE_DIR / "kb_tool.py") module = _load_module(BRIDGE_DIR / "kb_tool.py")
root = tmp_path / "workspace" root = tmp_path / "workspace"

View file

@ -9,9 +9,7 @@ MANIFEST = REPORT_DIR / "skill-pack-manifest.json"
def _skill(name: str) -> str: def _skill(name: str) -> str:
return (ROOT / ".agents" / "skills" / name / "SKILL.md").read_text( return (ROOT / ".agents" / "skills" / name / "SKILL.md").read_text(encoding="utf-8")
encoding="utf-8"
)
def test_manifest_indexes_valid_repo_native_skills() -> None: def test_manifest_indexes_valid_repo_native_skills() -> None:
@ -72,16 +70,19 @@ def test_gcp_skill_uses_current_operator_access_and_parity_truth() -> None:
"PostgreSQL 16.14", "PostgreSQL 16.14",
"public IP disabled", "public IP disabled",
"39/39", "39/39",
"52,164/52,164", "52,167/52,167",
"PID `148735`", "`148735`, `NRestarts=0`",
"NRestarts=0", "NRestarts=0",
"state.db", "state.db",
"leoclean-cloudsql-memory-sync.service", "leoclean-cloudsql-memory-sync.service",
"gcp-canonical-parity-live-20260712.json", "gcp-db-first-restore-current.json",
"gcp-cory-model-replay-current.json", "gcp-db-first-parity-current.json",
"gcp-operator-access-blocker-current.json", "gcp-db-first-blind-claim-current.json",
"gcp-db-first-cleanup-current.json",
"18/18",
"6/6",
"invalid_target", "invalid_target",
"rejected", "older staging copy",
): ):
assert required in text assert required in text
@ -105,10 +106,11 @@ def test_vps_onboarding_and_m3taversal_skills_point_to_current_proof() -> None:
assert "/opt/teleo-eval/.last-deploy-sha" in provenance assert "/opt/teleo-eval/.last-deploy-sha" in provenance
assert "does not prove a gateway restart" in " ".join(provenance.split()) assert "does not prove a gateway restart" in " ".join(provenance.split())
assert "gcp-cloud-sql-t3-live-readonly-current.md" in onboarding assert "gcp-cloud-sql-t3-live-readonly-current.md" in onboarding
assert "gcp-canonical-parity-live-20260712.json" in onboarding assert "gcp-db-first-working-leo-20260714.md" in onboarding
assert "gcp-cory-model-replay-current.json" in onboarding assert "gcp-db-first-parity-current.json" in onboarding
assert "working-leo-current-proof-20260712.md" in onboarding assert "gcp-db-first-blind-claim-current.json" in onboarding
assert "ssh teleo-gcp-staging" in onboarding assert "gcp-db-first-cleanup-current.json" in onboarding
assert "ssh -o BatchMode=yes teleo-gcp-staging true" in onboarding
assert "37/37" in outcomes assert "37/37" in outcomes
assert "Approved is not the same as applied" in outcomes assert "Approved is not the same as applied" in outcomes
assert "Address `@m3taversal` only as `m3taversal`, exactly" in outcomes assert "Address `@m3taversal` only as `m3taversal`, exactly" in outcomes

View file

@ -0,0 +1,374 @@
import argparse
import hashlib
import json
import subprocess
from pathlib import Path
import pytest
from ops import restore_gcp_generated_postgres_snapshot as restore
SERVICE_STATE = {
"ActiveState": "active",
"SubState": "running",
"MainPID": "148735",
"NRestarts": "0",
"ExecMainStartTimestamp": "Thu 2026-07-09 07:00:04 UTC",
}
def manifest_rows(database: str, *, target: bool, row_count: int = 1) -> list[dict]:
identity = {
"kind": "identity",
"database": database,
"server_version_num": 160014,
"transaction_read_only": "on",
"server_address": "10.61.0.3" if target else None,
"server_port": 5432 if target else None,
"ssl": target,
"ssl_version": "TLSv1.3" if target else None,
"database_collation": "en_US.UTF8" if target else "en_US.utf8",
"database_ctype": "en_US.UTF8" if target else "en_US.utf8",
"server_encoding": "UTF8",
}
rows = [
identity,
{"kind": "schemas", "items": ["kb_stage", "public"]},
{"kind": "extensions", "items": [{"name": "pgcrypto", "version": "1.3"}]},
{
"kind": "application_roles",
"items": [
{
"name": "kb_apply",
"can_login": True,
"superuser": False,
"create_db": False,
"create_role": False,
"inherit": True,
"replication": False,
"bypass_rls": False,
}
],
},
]
rows.extend(
{"kind": kind, "items": []}
for kind in (
"columns",
"constraints",
"indexes",
"sequences",
"views",
"functions",
"triggers",
"types",
"policies",
)
)
rows.extend(
[
{
"kind": "table",
"schema": "public",
"table": "claims",
"row_count": row_count,
"rowset_md5": "same" if row_count == 1 else "different",
},
{
"kind": "performance",
"query": "count_claims",
"elapsed_ms": 1.0,
"observed_rows": row_count,
},
]
)
return rows
def manifest_text(database: str, *, target: bool, row_count: int = 1) -> str:
return "".join(json.dumps(row) + "\n" for row in manifest_rows(database, target=target, row_count=row_count))
def args_for(tmp_path: Path, **overrides) -> argparse.Namespace:
dump = tmp_path / "teleo-canonical.dump"
dump.write_bytes(b"PGDMPfixture")
source_manifest = tmp_path / "source-manifest.jsonl"
source_manifest.write_text(manifest_text("teleo", target=False), encoding="utf-8")
values = {
"operation": "restore",
"execute": True,
"target_db": "teleo_clone_dbfirst_test",
"run_id": "gcp-restore-test12345",
"host": restore.DEFAULT_HOST,
"project": restore.DEFAULT_PROJECT,
"password_secret": restore.DEFAULT_SECRET,
"service": restore.DEFAULT_SERVICE,
"command_timeout": 2.0,
"psql_bin": "psql",
"run_root": tmp_path / "runs",
"dump": dump,
"expected_dump_sha256": hashlib.sha256(dump.read_bytes()).hexdigest(),
"source_manifest": source_manifest,
"manifest_sql": Path("ops/postgres_parity_manifest.sql").resolve(),
"pg_restore_bin": "pg_restore",
"restore_timeout": 5.0,
"manifest_timeout": 5.0,
"max_target_query_ms": 2000.0,
"max_target_source_ratio": 20.0,
}
values.update(overrides)
return argparse.Namespace(**values)
def install_restore_fakes(
monkeypatch: pytest.MonkeyPatch,
*,
target_manifest: str,
) -> dict[str, bool]:
state = {"exists": False, "dropped": False}
monkeypatch.setattr(restore.os, "geteuid", lambda: 0)
monkeypatch.setattr(restore, "service_state", lambda *_args, **_kwargs: dict(SERVICE_STATE))
monkeypatch.setattr(restore, "resolve_password", lambda *_args, **_kwargs: "private-password")
monkeypatch.setattr(
restore,
"validate_pg_restore_version",
lambda *_args, **_kwargs: {
"pg_restore_bin": "pg_restore",
"pg_restore_version": "pg_restore (PostgreSQL) 16.14",
"pg_restore_major": 16,
"source_postgres_major": 16,
"compatible": True,
},
)
monkeypatch.setattr(restore, "database_exists", lambda *_args, **_kwargs: state["exists"])
def fake_create(*_args, **_kwargs):
state["exists"] = True
def fake_drop(*_args, **_kwargs):
existed = state["exists"]
state["exists"] = False
state["dropped"] = True
return {"existed_before": existed, "clone_database_remaining": 0}
monkeypatch.setattr(restore, "create_database", fake_create)
monkeypatch.setattr(restore, "restore_dump", lambda *_args, **_kwargs: None)
monkeypatch.setattr(restore, "capture_target_manifest", lambda *_args, **_kwargs: target_manifest)
monkeypatch.setattr(
restore,
"rollback_database_state",
lambda *_args, **_kwargs: {"exists": True, "datallowconn": False, "connections": 0},
)
monkeypatch.setattr(restore, "drop_clone", fake_drop)
return state
def test_restore_success_retains_only_exact_parity_clone(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
args = args_for(tmp_path)
state = install_restore_fakes(
monkeypatch,
target_manifest=manifest_text(args.target_db, target=True),
)
result = restore.run_restore(args)
assert result["status"] == "pass"
assert result["phase"] == "retained_for_no_send_replay"
assert result["parity"]["problems"] == []
assert result["parity"]["details"]["source_total_rows"] == 1
assert result["target"]["connectivity"]["private_connectivity"] is True
assert result["target"]["connectivity"]["ssl"] is True
assert result["live_service"]["unchanged"] is True
assert result["cleanup"]["clone_database_remaining"] == 1
assert state == {"exists": True, "dropped": False}
run_dir = args.run_root / args.run_id
assert (run_dir / "target-manifest.jsonl").stat().st_mode & 0o777 == 0o600
assert (run_dir / "restore-receipt.json").stat().st_mode & 0o777 == 0o600
def test_parity_failure_drops_generated_clone_and_receipts_both_failures(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
args = args_for(tmp_path)
state = install_restore_fakes(
monkeypatch,
target_manifest=manifest_text(args.target_db, target=True, row_count=2),
)
result = restore.run_restore(args)
assert result["status"] == "fail"
assert result["error"]["phase"] == "parity_compare"
assert result["parity"]["problems"] == ["table_content_mismatches=1"]
assert result["cleanup"]["attempted"] is True
assert result["cleanup"]["clone_database_remaining"] == 0
assert state == {"exists": False, "dropped": True}
def test_create_command_failure_still_drops_a_committed_clone(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
args = args_for(tmp_path)
state = install_restore_fakes(
monkeypatch,
target_manifest=manifest_text(args.target_db, target=True),
)
def committed_then_failed(*_args, **_kwargs):
state["exists"] = True
raise restore.RestoreError("connection dropped after create committed")
monkeypatch.setattr(restore, "create_database", committed_then_failed)
result = restore.run_restore(args)
assert result["status"] == "fail"
assert result["error"]["phase"] == "database_create"
assert result["cleanup"]["attempted"] is True
assert result["cleanup"]["clone_database_remaining"] == 0
assert state == {"exists": False, "dropped": True}
def test_changed_live_service_turns_successful_restore_into_cleanup(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
args = args_for(tmp_path)
state = install_restore_fakes(
monkeypatch,
target_manifest=manifest_text(args.target_db, target=True),
)
calls = 0
def changing_service(*_args, **_kwargs):
nonlocal calls
calls += 1
result = dict(SERVICE_STATE)
if calls > 1:
result["NRestarts"] = "1"
return result
monkeypatch.setattr(restore, "service_state", changing_service)
result = restore.run_restore(args)
assert result["status"] == "fail"
assert result["error"]["phase"] == "service_and_rollback_readback"
assert result["live_service"]["unchanged"] is False
assert result["cleanup"]["clone_database_remaining"] == 0
assert state["dropped"] is True
def test_cleanup_is_bound_to_passing_restore_receipt(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
args = args_for(tmp_path)
state = install_restore_fakes(
monkeypatch,
target_manifest=manifest_text(args.target_db, target=True),
)
assert restore.run_restore(args)["status"] == "pass"
cleanup_args = argparse.Namespace(
**{
key: getattr(args, key)
for key in (
"target_db",
"run_id",
"host",
"project",
"password_secret",
"service",
"command_timeout",
"psql_bin",
"run_root",
)
},
operation="cleanup",
execute=True,
)
result = restore.run_cleanup(cleanup_args)
assert result["status"] == "pass"
assert result["database"] == {"existed_before": True, "clone_database_remaining": 0}
assert result["rollback_database"] == {"exists": True, "datallowconn": False, "connections": 0}
assert result["live_service"]["unchanged"] is True
assert state == {"exists": False, "dropped": True}
def test_restore_dump_uses_tls_no_owner_no_acl_and_keeps_secret_out_of_argv(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
args = args_for(tmp_path)
captured = {}
def fake_run(command, *, timeout, env=None, input_text=None):
captured.update(command=command, timeout=timeout, env=env, input_text=input_text)
return subprocess.CompletedProcess(command, 0, "", "")
monkeypatch.setattr(restore, "_run", fake_run)
restore.restore_dump(args, "private-password")
assert captured["command"][0] == "pg_restore"
assert captured["command"][1] == "-d"
assert "sslmode=require" in captured["command"][2]
assert "--no-owner" in captured["command"]
assert "--no-acl" in captured["command"]
assert "--exit-on-error" in captured["command"]
assert all("private-password" not in argument for argument in captured["command"])
assert captured["env"]["PGPASSWORD"] == "private-password"
def test_pg_restore_preflight_rejects_client_older_than_snapshot(
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
manifest_path = tmp_path / "source.jsonl"
manifest_path.write_text(manifest_text("teleo", target=False), encoding="utf-8")
source_manifest = restore.load_manifest(manifest_path)
def fake_run(command, *, timeout, env=None, input_text=None):
assert command == ["pg_restore", "--version"]
assert timeout > 0
assert env is None
assert input_text is None
return subprocess.CompletedProcess(command, 0, "pg_restore (PostgreSQL) 15.18\n", "")
monkeypatch.setattr(restore, "_run", fake_run)
with pytest.raises(restore.RestoreError, match="major 15 is older than source PostgreSQL major 16"):
restore.validate_pg_restore_version("pg_restore", source_manifest, timeout=2.0)
@pytest.mark.parametrize("target", ["teleo_canonical", "teleo_clone_UPPER", "other_clone_test"])
def test_parse_args_rejects_non_disposable_target(target: str) -> None:
with pytest.raises(SystemExit):
restore.parse_args(
[
"cleanup",
"--execute",
"--target-db",
target,
"--run-id",
"gcp-restore-test12345",
]
)
def test_parse_args_requires_explicit_execute() -> None:
with pytest.raises(SystemExit):
restore.parse_args(
[
"cleanup",
"--target-db",
"teleo_clone_test",
"--run-id",
"gcp-restore-test12345",
]
)