Retain installer failure results and map crashes to Core source

src/agent_comms/installer/launch.py

@@ -29,17 +29,21 @@
 from agent_comms.installer.refusals import UpgradeRefused
 from agent_comms.installer.commands import InstallCommands
 from agent_comms.installer.reporting import (
-    FinishedInstall, InstallOutcome, RootWriter, UnfinishedInstall, UnpostedWriter,
+    FailedInstall, FinishedInstall, InstallOutcome, ObservedInstall, RootWriter,
+    UnfinishedInstall, UnpostedWriter,
 )
+from agent_comms.installer.runtime_resolution import CorePackage
 from agent_comms.installer.probes.harnesses import HarnessesProbe
 from agent_comms.installer.upgrade import (
+    LiveRun,
     RehearsedUpgrade,
     Stack,
     TargetTree,
     Upgrade,
     upgrade,
 )
 from agent_comms.mro_dispatch import MroDispatch, handles
+from agent_comms.repo import Repo
 from agent_comms.storage_rules import StorageRule
 
 
@@ -50,6 +54,11 @@ class InstallReport(MroDispatch):
     writer: RootWriter = field(default_factory=UnpostedWriter)
     outcome: InstallOutcome = field(default_factory=UnfinishedInstall, init=False)
     bridges: SessionBridgeResumes = field(default_factory=SessionBridgeResumes, init=False)
+    source: Repo = field(init=False, repr=False)
+
+    def record_live(self, result: LiveRun) -> LiveRun:
+        self.outcome = ObservedInstall(result)
+        return result
 
     def complete(self, result: RehearsedUpgrade) -> RehearsedUpgrade:
         self.outcome = FinishedInstall(result)
@@ -65,22 +74,31 @@ def refused(self, error: UpgradeRefused) -> tuple[str, ...]:
 
     @handles(BaseException)
     def crashed(self, error: BaseException) -> tuple[str, ...]:
-        frame = extract_tb(error.__traceback__)[-1]
+        package = Path(agent_comms.__file__).resolve().parent
+        frame = next(
+            frame for frame in reversed(extract_tb(error.__traceback__))
+            if Path(frame.filename).resolve().is_relative_to(package)
+        )
+        module = ".".join(
+            Path(frame.filename).resolve().relative_to(package.parent).with_suffix("").parts
+        )
+        location = self.source.module_file(module)
         return (f"Install failed: {type(error).__name__}: {error}",
-                f"{frame.filename}:{frame.lineno}")
+                f"{location}:{frame.lineno}")
 
     @contextmanager
     def running(self, tree: TargetTree) -> Iterator[None]:
         """Every writer runs installed artifacts from outside ``tree``, so the report follows
         the tree's cleanup and a cleanup failure is the install's failure."""
+        self.source = Repo(checkout=tree.path, github=CorePackage.repository)
         try:
             with tree:
                 yield
             lines = self.outcome.lines()
         except BaseException as error:
-            failure = (*next(self.handlers_for(error))(error), *self.bridges.lines())
+            self.outcome = FailedInstall(self.outcome, next(self.handlers_for(error))(error))
             try:
-                self.writer.post(failure)
+                self.writer.post((*self.outcome.lines(), *self.bridges.lines()))
             except BaseException as publication_error:
                 error.add_note(f"Install failure report also failed: {publication_error}")
                 print(f"install: failure report also failed: {publication_error}", file=sys.stderr)

src/agent_comms/installer/reporting.py

@@ -9,18 +9,34 @@
 from agent_comms.installer.probes.report import InstallReportProbe
 
 if TYPE_CHECKING:
-    from agent_comms.installer.upgrade import RehearsedUpgrade, Stack
+    from agent_comms.installer.upgrade import LiveRun, RehearsedUpgrade, Stack
 
 
 class InstallOutcome(ABC):
     @abstractmethod
     def lines(self) -> tuple[str, ...]: ...
 
+    @abstractmethod
+    def live_lines(self) -> tuple[str, ...]: ...
+
 
 class UnfinishedInstall(InstallOutcome):
     def lines(self) -> tuple[str, ...]:
         raise RuntimeError("Install exited without producing its result")
 
+    def live_lines(self) -> tuple[str, ...]:
+        return ()
+
+
+@dataclass(frozen=True)
+class ObservedInstall(UnfinishedInstall):
+    """A live result returned before harness registration or cleanup can fail."""
+
+    live: LiveRun
+
+    def live_lines(self) -> tuple[str, ...]:
+        return self.live.lines()
+
 
 @dataclass(frozen=True)
 class FinishedInstall(InstallOutcome):
@@ -29,6 +45,23 @@ class FinishedInstall(InstallOutcome):
     def lines(self) -> tuple[str, ...]:
         return (self.result.finished, *self.result.lines())
 
+    def live_lines(self) -> tuple[str, ...]:
+        return self.result.live_lines()
+
+
+@dataclass(frozen=True)
+class FailedInstall(InstallOutcome):
+    """Failure retains the result already observed, without replacing its error."""
+
+    previous: InstallOutcome
+    failure: tuple[str, ...]
+
+    def lines(self) -> tuple[str, ...]:
+        return (*self.failure, *self.live_lines())
+
+    def live_lines(self) -> tuple[str, ...]:
+        return self.previous.live_lines()
+
 
 class RootWriter(ABC):
     @abstractmethod

src/agent_comms/installer/upgrade.py

@@ -348,6 +348,9 @@ class RehearsedUpgrade(Upgrade):
 
     finished: ClassVar[str] = "Install rehearsed"
 
+    def live_lines(self) -> tuple[str, ...]:
+        return ()
+
     def lines(self) -> tuple[str, ...]:
         slowest = sorted(self.commands, key=lambda timed: timed.seconds, reverse=True)[:5]
         return (*super().lines(),
@@ -368,8 +371,11 @@ class InstalledUpgrade(RehearsedUpgrade, Extendable):
     finished: ClassVar[str] = "Install finished"
     """The headline of the post that reports this upgrade."""
 
+    def live_lines(self) -> tuple[str, ...]:
+        return self.live.lines()
+
     def lines(self) -> tuple[str, ...]:
-        return (*super().lines(), *self.live.lines(), *self.processes.lines(),
+        return (*super().lines(), *self.live_lines(), *self.processes.lines(),
                 *(line for outcome in self.bridge_resumes for line in outcome.lines()))
 
 
@@ -819,7 +825,7 @@ def install(self, rehearsal: Rehearsed, bridges: SessionBridgeResumes,
             try:
                 for bridge in Serve.running(self.installed.home):
                     bridges.pause(bridge, resources)
-                return self.live(rehearsal, report)
+                return report.record_live(self.live(rehearsal, report))
             finally:
                 # A later pause may fail; resume captured peers before closing their sockets.
                 bridges.resume()