|
12 | 12 | import warnings |
13 | 13 | from dataclasses import dataclass, field |
14 | 14 | from enum import Enum |
| 15 | +from pathlib import Path |
15 | 16 | from typing import TYPE_CHECKING, Any, Optional |
16 | 17 |
|
17 | 18 | from teaagent.errors import ToolPermissionError |
@@ -110,6 +111,48 @@ class MultiSigQuorumConfig: |
110 | 111 | ) # Patterns triggering multi-sig |
111 | 112 | timeout_seconds: int = 300 # Timeout for collecting signatures |
112 | 113 |
|
| 114 | + @classmethod |
| 115 | + def from_workspace_config(cls, root: str | Path) -> MultiSigQuorumConfig: |
| 116 | + """Load ``multi_sig`` section from ``<root>/.teaagent/config.json``.""" |
| 117 | + from teaagent.config_loader import load_workspace_config |
| 118 | + |
| 119 | + section = load_workspace_config(root).get('multi_sig') |
| 120 | + if not isinstance(section, dict): |
| 121 | + return cls() |
| 122 | + |
| 123 | + peer_ids = section.get('peer_agent_ids') or [] |
| 124 | + if not isinstance(peer_ids, list): |
| 125 | + peer_ids = [] |
| 126 | + |
| 127 | + patterns = section.get('high_risk_patterns') or [] |
| 128 | + if not isinstance(patterns, list): |
| 129 | + patterns = [] |
| 130 | + |
| 131 | + relay_urls = section.get('peer_relay_urls') or {} |
| 132 | + if not isinstance(relay_urls, dict): |
| 133 | + relay_urls = {} |
| 134 | + |
| 135 | + public_keys = section.get('peer_public_keys') or {} |
| 136 | + if not isinstance(public_keys, dict): |
| 137 | + public_keys = {} |
| 138 | + |
| 139 | + local_relay = section.get('local_relay_base_url') |
| 140 | + local_relay_url = str(local_relay).strip() if local_relay else None |
| 141 | + if local_relay_url == '': |
| 142 | + local_relay_url = None |
| 143 | + |
| 144 | + return cls( |
| 145 | + enabled=bool(section.get('enabled', False)), |
| 146 | + required_approvals=int(section.get('required_approvals', 2)), |
| 147 | + peer_agent_ids=[str(item) for item in peer_ids], |
| 148 | + peer_public_keys={str(k): str(v) for k, v in public_keys.items()}, |
| 149 | + peer_relay_urls={str(k): str(v).rstrip('/') for k, v in relay_urls.items()}, |
| 150 | + local_relay_base_url=local_relay_url, |
| 151 | + allow_dev_signatures=bool(section.get('allow_dev_signatures', False)), |
| 152 | + high_risk_patterns=[str(item) for item in patterns], |
| 153 | + timeout_seconds=int(section.get('timeout_seconds', 300)), |
| 154 | + ) |
| 155 | + |
113 | 156 |
|
114 | 157 | @dataclass(frozen=True) |
115 | 158 | class ApprovalRequest: |
|
0 commit comments