Fix listener/TLS diagnostics and apply approved write access within active turns

amplifier_web/__init__.py

@@ -1,2 +1,2 @@
 """Amplifier's local web host."""
-__version__ = "0.69.1"
+__version__ = "0.69.2"

amplifier_web/chat_permissions.py

@@ -8,8 +8,8 @@
 from .shared_settings import overlay, read_settings
 
 WRITE_KEYS = ('allowed_write_paths', 'denied_write_paths')
-PENDING_DETAIL = ('Write settings are saved or being saved. Existing work keeps its current access until it finishes. '
-                  'Finish or stop existing child work before sending another message; no new turn has started.')
+PENDING_DETAIL = ('Write settings are saved or being saved. The live tool boundary has not confirmed the refresh. '
+                  'Finish or stop existing child work if refresh remains unavailable; no new turn has started.')
 
 
 class _Unchanged(Exception):
@@ -119,3 +119,31 @@ def edit(value):
         'This chat already inherits workspace defaults.' if reset else
         'Chat write overrides saved.' if changed else 'Chat write overrides are unchanged.'))
     return result
+
+
+async def refresh_active(management, session, saved_revision, execution_epoch):
+    """Release only this write-change fence after the same worker confirms it."""
+    service = management.service
+    refresh = getattr(service.runtime, 'refresh_file_access', None)
+    if not callable(refresh):
+        return
+    async with service.lock:
+        if (not session.get('fileAccessPending') or session.get('configurationBusy')
+                or session.get('configurationPendingRevision', 0) != execution_epoch
+                or snapshot(management, session)['revision'] != saved_revision):
+            return
+        revision = session.get('configurationPendingRevision', 0)
+        settings = read_settings(session['workspace'], shared_home=management.settings.shared_home,
+                                 session_id=native_id(session))
+        settings_revision = fingerprint(settings)
+    if not await refresh(session['id'], settings_revision):
+        return
+    async with service.lock:
+        if (session.get('configurationBusy') or not session.get('fileAccessPending')
+                or session.get('configurationPendingRevision', 0) != revision
+                or snapshot(management, session)['revision'] != saved_revision):
+            return
+        session['configurationPending'] = False
+        session.pop('fileAccessPending', None)
+        session.pop('configurationRefresh', None)
+        service._publish_changes(sessions={session['id']})

amplifier_web/file_source_guard.py

@@ -162,7 +162,15 @@ async def execute(self, tool, kind, arguments):
         from amplifier_core import ToolResult
         targets = {}
         invoked = False
+        original = tool
+        policy = self.coordinator.get_capability('web.write_policy')
+        def invocation():
+            if policy and kind != 'read':
+                module = 'tool-apply-patch' if kind == 'patch' else 'tool-filesystem'
+                return policy.invocation(original, module, self.coordinator)
+            return original
         try:
+            tool = invocation()
             targets = self._targets(tool, kind, arguments)
             # Capture before waiting: a simultaneous write by THIS coordinator
             # must not silently inherit a version produced while it waited.
@@ -172,6 +180,7 @@ async def execute(self, tool, kind, arguments):
                     expected[path] = None
             async with _locked(self.directory, targets):
                 # Re-resolve aliases and recheck policy under admission.
+                tool = invocation()
                 if targets != self._targets(tool, kind, arguments):
                     raise SourceConflict('file_source_stale', 'The file path changed. Read it again before editing.')
                 if kind == 'read':
@@ -252,8 +261,11 @@ async def install_file_source_guard(coordinator, directory=None):
         return
     from .host.config import app_home
     guard = FileSourceGuard(coordinator, directory or app_home() / 'file-source-locks')
+    from .write_policy_refresh import IMAGES, PolicyImageTool
     for name, tool in list((coordinator.get('tools') or {}).items()):
         kind = _KNOWN.get((type(tool).__module__, type(tool).__name__))
         if kind:
             await coordinator.mount('tools', GuardedFileTool(tool, guard, kind), name=name)
+        elif (type(tool).__module__, type(tool).__name__) in IMAGES:
+            await coordinator.mount('tools', PolicyImageTool(tool, coordinator), name=name)
     coordinator.register_capability('web.file_source_guard', guard)

amplifier_web/host/config.py

@@ -360,6 +360,16 @@ def load_config(workspace, *, home=None, legacy_home=None, session_id=None, glob
     return read_config(workspace, home=home, shared_home=shared, session_id=session_id, global_only=global_only, owner_session_id=owner_session_id)
 
 
+def runtime_settings(settings, home, registry_home):
+    """Normalize canonical settings for the selected runtime snapshot."""
+    def relocate(value):
+        if isinstance(value, dict): return {k:relocate(v) for k,v in value.items()}
+        if isinstance(value, list): return [relocate(v) for v in value]
+        if isinstance(value, str): return value.replace(str(home / "foundation"), str(registry_home))
+        return value
+    return relocate(settings)
+
+
 def read_config(workspace, *, home=None, shared_home=None, session_id=None, global_only=False, settings_cache=None, registry_home=None, owner_session_id=None):
     """Read runtime settings without loading keys, preparing caches or writing."""
     if os.name == "nt":
@@ -374,14 +384,13 @@ def read_config(workspace, *, home=None, shared_home=None, session_id=None, glob
     from ..updates import foundation_home
     registry_home = Path(registry_home) if registry_home is not None else foundation_home(home)
     # Keep explicit app-cache source overrides aligned with the active snapshot.
-    def relocate(value):
-        if isinstance(value, dict): return {k:relocate(v) for k,v in value.items()}
-        if isinstance(value, list): return [relocate(v) for v in value]
-        if isinstance(value, str): return value.replace(str(home / "foundation"), str(registry_home))
-        return value
     from ..access_mode import read_mode
     from ..image_migration import scoped_image_sources
-    return HostConfig(home, workspace, relocate(settings), registry_home, shared, global_only,
+    result = HostConfig(home, workspace, runtime_settings(settings, home, registry_home), registry_home, shared, global_only,
                       read_mode(home, workspace, session_id, owner_session_id=owner_session_id),
                       tuple(scoped_image_sources(workspace, shared_home=shared,
                                                 session_id=session_id, global_only=global_only)))
+    # Host provenance for settings-only refresh comparisons. Relocated runtime
+    # source paths must not look like a concurrent canonical settings change.
+    result._canonical_settings = copy.deepcopy(settings)
+    return result

amplifier_web/host/session.py

@@ -742,6 +742,8 @@ async def compose_configured_bundle(registry, loaded, config, *, execution_works
         # Routing is now an ordinary opt-out-capable default. A retained matrix
         # setting must not silently restore a behavior the user disabled.
         explicit_context_forwarding(loaded, config.settings, context_defaults)
+        from ..write_policy_refresh import WritePolicySource
+        write_policy_source = WritePolicySource(loaded, config, execution_workspace=execution_workspace)
         loaded = _apply_settings(loaded, config)
         # Existing root defaults are explicit selections, not a new forced
         # version. Reuse their effective source for child/sibling declarations.
@@ -754,6 +756,11 @@ async def compose_configured_bundle(registry, loaded, config, *, execution_works
     for key, value in root_identity.items():
         setattr(loaded, key, value)
     loaded = components.apply(loaded)
+    if snapshot:
+        from ..write_policy_refresh import WritePolicySource
+        write_policy_source = WritePolicySource(loaded, config, snapshot=True,
+                                               execution_workspace=execution_workspace)
+    loaded._host_write_policy_source = write_policy_source
     # Credentials and safety policy remain local host responsibilities. They
     # do not add modules or replace the saved source/model/routing selections.
     loaded = _apply_host_policy(loaded, config, execution_workspace=execution_workspace)
@@ -889,6 +896,11 @@ def apply_runtime_plan(loaded, edited, config, execution_workspace):
     for key in ("session", "agents", "context", "instruction"):
         if key in edited:
             setattr(loaded, key, edited[key])
+    from ..write_policy_refresh import WritePolicySource
+    # An edited live plan is already composed. Retain its explicit ceilings,
+    # rather than trusting provenance imported from bundle data.
+    loaded._host_write_policy_source = WritePolicySource(loaded, config, snapshot=True,
+                                                        execution_workspace=execution_workspace)
     loaded = _apply_host_policy(loaded, config, execution_workspace=execution_workspace)
     for key in ("providers", "tools", "hooks", "session", "agents"):
         if key in edited:
@@ -1165,6 +1177,9 @@ def retain_session(constructing):
                 **initialization)
         decision_accounts.require_complete()
         coordinator = session.coordinator
+        source = getattr(loaded, '_host_write_policy_source', None)
+        if source is not None:
+            coordinator.register_capability('web.write_policy.source', source)
         coordinator.register_capability("approval.interactive", approvals.interactive)
         from ..read_policy import validate_mounted_read_policy
         validate_mounted_read_policy(coordinator)

amplifier_web/installation_health.py

@@ -14,7 +14,7 @@
 def doctor(home):
     from .deployment import load_server_config
     import yaml
-    from .tls import ssl_context
+    from .tls import certificate_checks, ssl_context
     checks = []
     from .host_dependencies import MissingHostDependency, require_uv, tool_path
     import shutil
@@ -49,7 +49,8 @@ def doctor(home):
     try:
         ssl_context(home, config)
         checks.append({"ok": True, "message": "Secure connection is configured" if config['tls']['method'] != 'none' else "Local connection is configured"})
-    except (OSError, ValueError) as error:
+        checks.extend(certificate_checks(home, config))
+    except (OSError, ValueError, NotImplementedError) as error:
         checks.append({"ok": False, "message": "The connection certificate needs attention. Check your TLS settings.", "detail": str(error)})
     return {"ok": all(row['ok'] for row in checks), "version": __version__, "checks": checks}
 
@@ -123,7 +124,8 @@ async def _status_once(home, wait=0):
     try:
         config = load_server_config(home)
         manager = SimpleNamespace(home=home, service=SimpleNamespace(server_config=config, port=config['port']))
-        urls, pin = probe_targets(manager)
+        scope = {}
+        urls, pin = await asyncio.to_thread(probe_targets, manager, diagnostics=scope)
         token = (Path(home) / 'config/auth/control-token').read_text().strip()
     except (OSError, ValueError, yaml.YAMLError) as error:
         if initial_state == 'not configured':
@@ -132,34 +134,48 @@ async def _status_once(home, wait=0):
         return {"ok": False, "service": initial_state, "message": "Unified's connection settings need attention. Run amplifier-unified doctor.", "detail": str(error)}
     deadline = asyncio.get_running_loop().time() + wait
     state = 'unknown'
+    listener_checks = []
+    diagnostics = {'configuredBinds': config['bind'], 'publicUrls': config['public_origins'],
+                   'publicUrlsVerified': False, **scope}
+    slots = asyncio.Semaphore(4)
     async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=2), trust_env=False) as client:
-        while True:
+        async def check_listener(url):
+            row = {'url': url.removesuffix('/api/health'), 'verified': False}
             try:
-                state = await asyncio.to_thread(service_state, home)
-            except (OSError, RuntimeError, subprocess.SubprocessError):
-                state = 'unavailable'
-            for url in urls:
-                try:
+                async with slots:
                     async with client.get(url, ssl=pin, allow_redirects=False,
                                           headers={'Authorization': 'Bearer ' + token, 'Host': f"localhost:{config['port']}"}) as response:
                         value = await response.json() if response.status == 200 else None
                         if matches_health(value, home, expected):
-                            # A reachable manual host is useful information, but
-                            # cannot stand in for a healthy managed service.
-                            ok = state == 'running'
-                            if not ok and asyncio.get_running_loop().time() < deadline:
-                                continue
-                            return {"ok": ok, "responding": True, "service": state,
-                                    "message": "Unified is running and responding on this computer." if ok else "Unified responds on this computer, but its background service is " + state + ".",
-                                    "url": url.removesuffix('/api/health'),
-                                    "publicUrls": config['public_origins'],
-                                    "publicUrlsVerified": False}
-                except (aiohttp.ClientError, OSError, TimeoutError, ValueError):
-                    pass
+                            row['verified'] = True
+                        else:
+                            row['detail'] = 'No matching authenticated installation health response.'
+            except (aiohttp.ClientError, OSError, TimeoutError, ValueError):
+                # Do not print exception text that could include credentials.
+                row['detail'] = 'Connection, TLS pin, or health response failed.'
+            return row
+
+        while True:
+            try:
+                state = await asyncio.to_thread(service_state, home)
+            except (OSError, RuntimeError, subprocess.SubprocessError):
+                state = 'unavailable'
+            # Resolve once per pass; connect numerically with at most four
+            # concurrent, two-second requests. One failed bind cannot hide others.
+            listener_checks = await asyncio.gather(*(check_listener(url) for url in urls))
+            verified = [row['url'] for row in listener_checks if row['verified']]
+            if verified and (state == 'running' or asyncio.get_running_loop().time() >= deadline):
+                # A manual host cannot stand in for a healthy managed service.
+                ok = state == 'running'
+                return {"ok": ok, "responding": True, "service": state,
+                        "message": "Unified is running and responding on this computer." if ok else "Unified responds on this computer, but its background service is " + state + ".",
+                        "url": verified[0], "listenerUrls": verified,
+                        "listenerChecks": listener_checks, **diagnostics}
             if asyncio.get_running_loop().time() >= deadline:
                 break
             await asyncio.sleep(.5)
     return {"ok": False, "responding": False, "service": state,
+            "listenerUrls": [], "listenerChecks": listener_checks, **diagnostics,
             "message": f"Unified is not ready yet (background service: {state}).",
             "nextStep": "Run amplifier-unified service logs to see why startup has not completed."}
 
@@ -176,8 +192,14 @@ def display(result, *, as_json=False, verbose=False):
                 print('  ' + row['detail'])
     else:
         print(result['message'])
-        if result.get('ok') and result.get('url'):
-            print('Host listener: ' + result['url'])
+        for url in result.get('listenerUrls', [result['url']] if result.get('ok') and result.get('url') else []):
+            print('Host listener: ' + url)
+        for row in result.get('listenerChecks', []):
+            if not row['verified']:
+                print('Listener not verified: ' + row['url'] + ' — ' + row['detail'])
+        if result.get('configuredBinds'):
+            print('Configured binds (not listener proof): ' + ', '.join(result['configuredBinds']))
+            print('Probe scope: ' + result['listenerScope'])
         for url in result.get('publicUrls', []):
             print('Configured public address (not checked): ' + url)
         if result.get('publicUrls'):