Prepare shared component generations and adopt updates per worker

amplifier_web/__init__.py

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

amplifier_web/bundles.py

@@ -176,6 +176,19 @@ def sanitize_export(value, path=(), secrets=None):
     return copy.deepcopy(value)
 
 
+def offered_profiles(config, registry=None):
+    """The picker and updater share one current, app-level profile authority."""
+    if registry is None:
+        path = config.registry_home / 'registry.json'
+        registry = json.loads(path.read_text()).get('bundles', {}) if path.exists() else {}
+    available = set(registry) | set(config.registrations)
+    names = (STANDALONE_PROFILES & available) | set(config.settings.get('bundle', {}).get('added', {}))
+    entries = BundleManager.entries(config.settings)
+    disabled = {row.get('name') for row in entries
+                if row.get('role') == 'standalone' and row.get('enabled') is False}
+    return sorted(names - disabled)
+
+
 class BundleManager:
     def __init__(self, home: Path, *, store=None):
         if store is None:
@@ -345,13 +358,9 @@ def register(current):
                 # describe an add-on, so neither cache flag admits picker rows.
                 # Source overrides also name dependencies; they are not standalone
                 # registrations. Keep the capability catalog below unfiltered.
-                available = set(registry) | set(config.registrations)
-                names = STANDALONE_PROFILES & available
-                names.update(config.settings.get('bundle', {}).get('added', {}))
-
-                disabled = {row['name'] for row in self.entries(settings) if row.get('role')=='standalone' and row.get('enabled') is False}
+                names = offered_profiles(config, registry)
                 from .bundle_selection import catalog_entry
-                catalog = sorted((catalog_entry(name, catalog_metadata(config, registry, name)) for name in names-disabled),
+                catalog = sorted((catalog_entry(name, catalog_metadata(config, registry, name)) for name in names),
                                  key=lambda row: (row['label'].casefold(), row['name'].casefold(), row['name']))
                 return {"bundles": self.public_entries(settings), "registeredBundles": catalog}
             def mutate(current):

amplifier_web/generation_leases.py

@@ -0,0 +1,39 @@
+"""Private durable references from live worker processes to source generations."""
+
+from __future__ import annotations
+
+import json
+import os
+import uuid
+from pathlib import Path
+
+
+def acquire(home, generation, pid):
+    from .host.config import write_private
+
+    path = Path(home) / "updates/worker-leases" / (uuid.uuid4().hex + ".json")
+    write_private(path, json.dumps({"schema": 1, "generation": generation, "pid": pid}))
+    return path
+
+
+def references(home):
+    generations = set()
+    for path in (Path(home) / "updates/worker-leases").glob("*.json"):
+        record = json.loads(path.read_text())
+        if (
+            record.get("schema") != 1
+            or type(record.get("pid")) is not int
+            or record["pid"] <= 0
+        ):
+            raise ValueError("Unrecognized worker lease; storage retained")
+        try:
+            os.kill(record["pid"], 0)
+        except ProcessLookupError:
+            # Crash recovery: no live holder remains. PID reuse can only retain
+            # too much storage, never remove a live generation.
+            path.unlink(missing_ok=True)
+            continue
+        except PermissionError:
+            pass
+        generations.add(record["generation"])
+    return generations

amplifier_web/host/config.py

@@ -226,14 +226,14 @@ 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)
 
 
-def read_config(workspace, *, home=None, shared_home=None, session_id=None, global_only=False, settings_cache=None):
+def read_config(workspace, *, home=None, shared_home=None, session_id=None, global_only=False, settings_cache=None, registry_home=None):
     """Read runtime settings without loading keys, preparing caches or writing."""
     home = Path(home or app_home()).expanduser().resolve()
     workspace = Path(workspace).expanduser().resolve(strict=True)
     shared = Path(shared_home or amplifier_home()).expanduser().resolve()
     settings = read_settings(workspace, shared_home=shared, session_id=session_id, global_only=global_only, cache=settings_cache)
     from ..updates import foundation_home
-    registry_home = foundation_home(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()}

amplifier_web/host/session.py

@@ -567,30 +567,50 @@ def take(self, config, bundle, *, execution_workspace=None):
         return root
 
 
-async def prepare_dependencies(workspace, *, bundle=None, install_overrides=None):
+def apply_runtime_plan(loaded, edited, config, execution_workspace):
+    from ..runtime_controls import validate_plan
+    validate_plan(edited)
+    for key in ("providers", "tools", "hooks"):
+        if key in edited:
+            setattr(loaded, key, [{k:v for k,v in row.items() if k != "enabled"}
+                for row in edited[key] if row.get("enabled", True)])
+    for key in ("session", "agents", "context", "instruction"):
+        if key in edited:
+            setattr(loaded, key, edited[key])
+    loaded = _apply_host_policy(loaded, config, execution_workspace=execution_workspace)
+    for key in ("providers", "tools", "hooks", "session", "agents"):
+        if key in edited:
+            setattr(loaded, key, _expand_module_configuration(getattr(loaded, key)))
+    return loaded
+
+
+async def prepare_dependencies(workspace, *, bundle=None, install_overrides=None, dependency_batch=None, global_only=False, runtime_plan=None):
     """Prepare a fresh qualification worker before importing its live runtime.
 
     This first probe only installs configured dependencies. A separate process
     mounts the resulting frozen worker normally; source changes never mix an
     already imported wheel runtime with an editable cache in one interpreter.
     No conversation, history store, job recovery or model execution is opened.
     """
-    config = load_config(workspace)
+    config = load_config(workspace, global_only=global_only)
     from .config import prepare_registry
     prepare_registry(config)
     execution_workspace = Path(config.workspace).expanduser().resolve(strict=True)
     os.chdir(execution_workspace)
     _, loaded, _ = await load_root_bundle(config, bundle or config.active_bundle,
                                           execution_workspace=execution_workspace)
     snapshot = is_snapshot(loaded)
+    if runtime_plan is not None:
+        loaded = apply_runtime_plan(loaded, runtime_plan, config, execution_workspace)
     adapted, _ = live_plan(loaded.to_mount_plan())
     components = getattr(loaded, '_host_components', None) or required_components()
     loaded.session = adapted['session']
     loaded.agents = adapted.get('agents', {})
     components.apply(loaded)
     from .config import configure_skill_cache
     configure_skill_cache(loaded, config.registry_home)
-    await loaded.prepare(strict=True, refresh_dependencies=True,
+    return await loaded.prepare(strict=True, refresh_dependencies=True,
+        **({"dependency_batch": dependency_batch} if dependency_batch is not None else {}),
         **({'install_overrides': Path(install_overrides)} if install_overrides is not None else {}),
         cache_dir=config.registry_home / 'cache',
         source_resolver=lambda module, source: module_source(config, snapshot, module, source, components))
@@ -601,7 +621,7 @@ async def prepare_manager(workspace, *, runtime=None, bundle=None, background_de
                           application_host="Amplifier Unified", shared_handle=None,
                           shared_handle_getter=None, shared_snapshot=None,
                           write_guard=None, resolved_root=None, execution_workspace=None,
-                          refresh_dependencies=False, install_overrides=None, qualification_readonly=False, **kwargs):
+                          refresh_dependencies=False, install_overrides=None, qualification_readonly=False, global_only=False, runtime_plan=None, **kwargs):
     if refresh_dependencies and resume:
         raise ValueError("Dependency refresh is limited to a new isolated qualification session.")
     from amplifier_foundation import SessionConfigurator
@@ -613,7 +633,7 @@ async def prepare_manager(workspace, *, runtime=None, bundle=None, background_de
     from .storage import SessionStore
 
     runtime = runtime or Runtime()
-    config = load_config(workspace, session_id=runtime.session_id)
+    config = load_config(workspace, session_id=runtime.session_id, global_only=global_only)
     from .config import prepare_registry
     prepare_registry(config)
     # Registry/cache ownership is passed explicitly below. AMPLIFIER_HOME must
@@ -663,22 +683,11 @@ def legacy_value(name, default=None):
     registry, loaded, chosen = (resolved_root.take(config, chosen, execution_workspace=execution_workspace) if resolved_root is not None
                                else await load_root_bundle(config, chosen, execution_workspace=execution_workspace))
     snapshot = is_snapshot(loaded)
-    from ..runtime_controls import override_path, validate_plan
+    from ..runtime_controls import override_path
     edited_path = override_path(runtime.session_id)
-    if edited_path.exists():
-        edited = json.loads(edited_path.read_text())
-        validate_plan(edited)
-        for key in ("providers", "tools", "hooks"):
-            if key in edited:
-                setattr(loaded, key, [{k:v for k,v in row.items() if k != "enabled"}
-                    for row in edited[key] if row.get("enabled", True)])
-        for key in ("session", "agents", "context", "instruction"):
-            if key in edited:
-                setattr(loaded, key, edited[key])
-        loaded = _apply_host_policy(loaded, config, execution_workspace=execution_workspace)
-        for key in ("providers", "tools", "hooks", "session", "agents"):
-            if key in edited:
-                setattr(loaded, key, _expand_module_configuration(getattr(loaded, key)))
+    if runtime_plan is not None or edited_path.exists():
+        edited = runtime_plan if runtime_plan is not None else json.loads(edited_path.read_text())
+        loaded = apply_runtime_plan(loaded, edited, config, execution_workspace)
     if not snapshot:
         from .mentions import include_instruction_files
         loaded = include_instruction_files(loaded)
@@ -712,7 +721,7 @@ def progress(action, detail):
         preparation_policy["install_overrides"] = Path(install_overrides)
     # Activate modules from the same generation as the bundle registry. The
     # shared AMPLIFIER_HOME still owns history/settings, not app module caches.
-    prepared = await loaded.prepare(strict=True, install_deps=not qualification_readonly, **preparation_policy,
+    prepared = await loaded.prepare(strict=True, install_deps=not (qualification_readonly or os.environ.get("AMPLIFIER_RUNTIME_IMMUTABLE") == "1"), **preparation_policy,
         cache_dir=config.registry_home / "cache",
         source_resolver=lambda module, source: module_source(config, snapshot, module, source, components),
         progress_callback=progress)

amplifier_web/release-notes.json

@@ -1,6 +1,25 @@
 {
   "schemaVersion": 1,
   "releases": [
+    {
+      "version": "0.20.46",
+      "title": "Prepare shared component generations and adopt updates between work",
+      "changes": [
+        "Prepare the currently offered app profiles together in one dependency installation, then check each profile in a fresh process. Historical chats and workspace combinations no longer multiply routine update preparation.",
+        "Reuse indexed source inventories and coalesced availability checks. Proven catalog-data-only changes reuse a qualified runtime; code, dependency and configuration changes still receive isolated qualification.",
+        "Share qualified component environments and exact source revisions across compatible conversation workers. Custom configurations prepare separately when first used, and conflicting requirements stop a candidate without changing the running environment.",
+        "Make compatible component updates available to new workers immediately. Active workers finish on their existing generation and adopt the update after safe retirement; active voice, queued input, approvals and uncertain operations remain protected.",
+        "Retain rollback and referenced generations, and reclaim only proven unused owned environments. Application updates still use the coordinated restart boundary. CLI caches and registries remain separate, and native history is preserved without replay."
+      ],
+      "notices": [
+        {
+          "id": "shared-component-generations",
+          "title": "Use shared component generations",
+          "detail": "Component updates now prepare the current app offerings together. Already running conversation workers keep their loaded version until their work can safely retire.",
+          "action": "Let the included-components update finish once after installing this release. New work then uses the qualified generation automatically; application updates still require their normal restart."
+        }
+      ]
+    },
     {
       "version": "0.20.45",
       "title": "Save changed conversation records and reuse unaffected client views",