bingshuo
8485822da6
Part 2/4 of verified local 18dfdfd: fail-closed navigation and intent gate, fixed Agent connector, loopback code-channel candidate scripts and tests.
127 lines
3.8 KiB
Python
Executable file
127 lines
3.8 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
"""Read, acknowledge, restore intent, unlock one Agent, then call the fixed connector."""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
|
|
BASE_URL = "http://127.0.0.1:8031"
|
|
CONNECTOR = "/usr/local/sbin/guanghu-fixed-agent-connector"
|
|
|
|
|
|
def parse_args():
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--intent-state", required=True)
|
|
parser.add_argument("--principal-id", required=True)
|
|
parser.add_argument("--subject-kind", required=True, choices=("human", "persona", "persona_system"))
|
|
parser.add_argument("--human-anchor", required=True)
|
|
parser.add_argument("--agent-id", required=True)
|
|
parser.add_argument("--action", required=True)
|
|
parser.add_argument("--target-node-id", required=True)
|
|
parser.add_argument("--execute", action="store_true")
|
|
return parser.parse_args()
|
|
|
|
|
|
def request_json(path, token="", headers=None, payload=None):
|
|
request_headers = dict(headers or {})
|
|
if token:
|
|
request_headers["X-Lighthouse-Admin-Token"] = token
|
|
data = None
|
|
method = "GET"
|
|
if payload is not None:
|
|
request_headers["Content-Type"] = "application/json"
|
|
data = json.dumps(payload).encode()
|
|
method = "POST"
|
|
request = urllib.request.Request(
|
|
BASE_URL + path,
|
|
data=data,
|
|
method=method,
|
|
headers=request_headers,
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=10) as response:
|
|
return json.load(response)
|
|
except urllib.error.HTTPError as error:
|
|
print("Gate rejected %s with HTTP %d." % (path, error.code), file=sys.stderr)
|
|
raise SystemExit(4)
|
|
except (OSError, ValueError) as error:
|
|
print("Gate request failed: %s" % error, file=sys.stderr)
|
|
raise SystemExit(4)
|
|
|
|
|
|
def main():
|
|
args = parse_args()
|
|
token = os.environ.get("LIGHTHOUSE_ADMIN_TOKEN", "")
|
|
if not token:
|
|
print("LIGHTHOUSE_ADMIN_TOKEN is required in the process environment.", file=sys.stderr)
|
|
raise SystemExit(3)
|
|
with open(args.intent_state, encoding="utf-8") as handle:
|
|
capsule = json.load(handle)
|
|
identity_headers = {
|
|
"X-Guanghu-Principal-Id": args.principal_id,
|
|
"X-Guanghu-Subject-Kind": args.subject_kind,
|
|
"X-Guanghu-Human-Anchor": args.human_anchor,
|
|
}
|
|
navigation = request_json("/v2/global-navigation-map")
|
|
acknowledged = request_json(
|
|
"/v1/navigation-map/ack",
|
|
token=token,
|
|
headers=identity_headers,
|
|
payload={"map_hash": navigation["map_hash"]},
|
|
)
|
|
if not acknowledged.get("ok"):
|
|
raise SystemExit(4)
|
|
restored = request_json(
|
|
"/v1/intent-state/restore",
|
|
token=token,
|
|
headers=identity_headers,
|
|
payload=capsule,
|
|
)
|
|
if not restored.get("ok"):
|
|
raise SystemExit(4)
|
|
unlocked = request_json(
|
|
"/v1/agents/unlock",
|
|
token=token,
|
|
headers=identity_headers,
|
|
payload={
|
|
"agent_id": args.agent_id,
|
|
"action": args.action,
|
|
"target_node_id": args.target_node_id,
|
|
},
|
|
)
|
|
unlock_id = unlocked.get("unlock_id", "")
|
|
if not unlock_id:
|
|
raise SystemExit(4)
|
|
if not args.execute:
|
|
print(unlock_id)
|
|
return
|
|
completed = subprocess.run(
|
|
[
|
|
CONNECTOR,
|
|
"--unlock-id",
|
|
unlock_id,
|
|
"--principal-id",
|
|
args.principal_id,
|
|
"--subject-kind",
|
|
args.subject_kind,
|
|
"--human-anchor",
|
|
args.human_anchor,
|
|
"--agent-id",
|
|
args.agent_id,
|
|
"--action",
|
|
args.action,
|
|
"--target-node-id",
|
|
args.target_node_id,
|
|
],
|
|
check=False,
|
|
)
|
|
raise SystemExit(completed.returncode)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|