Safe capture lifecycle and capability contract

CLAUDE.md

@@ -37,7 +37,7 @@ Spark event log (JSONL)
 | `sparkparse/parse.py`   | `get_parsed_metrics()` is the main entry point. `parse_spark_ui_tree()` converts indented ASCII plans to node graphs. `parse_log()` orchestrates everything.     |
 | `sparkparse/clean.py`   | `log_to_dag_df()` and `log_to_combined_df()` produce the two output DataFrames. `get_readable_size()` and `get_readable_timing()` are Polars expression helpers. |
 | `sparkparse/app.py`     | Typer CLI. `get` → parses and writes output files. `viz` → launches dashboard.                                                                                   |
-| `sparkparse/capture.py` | `SparkparseCapture` context manager/decorator. Stops the active session, restarts it with event logging enabled, then processes logs on `__exit__`.              |
+| `sparkparse/capture.py` | `SparkparseCapture` context manager/decorator. Borrows configured sessions, supports explicit owned sessions, and finalizes results on `__exit__`.              |
 
 ## Data model
 
@@ -179,8 +179,8 @@ uv run pyrefly check sparkparse/ tests/  # type check
 
 ## Known quirks
 
-- `capture.py` stops and restarts the SparkSession to enable event logging — this resets
-  all in-memory cached DataFrames. Warn users accordingly.
+- `capture.py` borrows supplied SparkSessions and requires event logging to be enabled
+  before capture. Use `cap.spark` inside the context; opt into an owned session explicitly.
 - `test.py` and `test_capture.py` in `tests/` are integration tests that spin up a local
   SparkSession. They are slow and cannot run in CI without a JVM available.
 - `falsa` is listed in `[project.dependencies]` but its usage in the codebase is unclear —

Justfile

@@ -1,4 +1,8 @@
-export JAVA_HOME := `/usr/libexec/java_home -v 17 2>/dev/null || echo /opt/homebrew/opt/openjdk@17`
+# Prefer a JAVA_HOME that actually has a java binary in it: CI sets a good one
+# with setup-java, and a shell profile can export one left behind by an
+# upgraded JDK. Fall back to the macOS lookup when neither holds.
+env_java := env_var_or_default("JAVA_HOME", "")
+export JAVA_HOME := if path_exists(env_java / "bin" / "java") == "true" { env_java } else { `/usr/libexec/java_home -v 17 2>/dev/null || echo /opt/homebrew/opt/openjdk@17` }
 
 ci:
     uv run ruff check sparkparse/ tests/

README.md

@@ -42,9 +42,30 @@ import sparkparse
 with sparkparse.capture_context(spark=spark, action="get") as cap:
     df.groupBy("id").count().show()
 
-parsed = cap._parsed_logs  # ParsedLogDataFrames with .dag and .combined
+parsed = cap.dfs  # ParsedLogDataFrames with .dag and .combined
+result = cap.result  # dataframes, metadata, capabilities, and diagnostics
 ```
 
+Capture borrows a supplied Spark session and never stops or recreates it. Borrowed
+classic sessions must have event logging enabled before the workload starts. When
+sparkparse should own a local session, opt in explicitly with
+`capture_context(own_session=True)`. Connect captures expose the same result contract,
+with unavailable task/stage telemetry called out in `result.capabilities` rather than
+represented as zeroes.
+
+Use `backend="classic"` or `backend="connect"` to override detection. For post-run
+ingestion without a Spark session, use `backend="event_log", log_file="/path/to/log"`.
+Borrowed captures select the current application's log; use an explicit `log_file`
+when its filename cannot be identified. They report ambiguous capture scope and
+may observe an incomplete log. Owned captures stop their session before parsing.
+
+Capture failures raise by default; `capture_errors="record"` retains diagnostics
+and a partial result instead. This is independent of parser `strict=True`. Failed
+captures retain their temporary logs. User workload exceptions always take precedence.
+`action="viz"` produces an in-memory HTML report in `cap.report`, including coverage,
+on either backend; it does not launch a dashboard server. Capture-result JSON uses
+Arrow IPC tables to preserve schemas, including empty tables and nested metrics.
+
 ### decorator
 
 ```python

notebooks/debug_serverless_node_types.py

@@ -61,16 +61,26 @@
     display(exploded.limit(20))
     print(f"exploded row count: {exploded.count():,}")
 
-dfs = cap._parsed_logs
+dfs = cap.dfs
 assert dfs is not None
+assert cap.result is not None
+assert cap.result.metadata.backend == "spark_connect"
+assert cap.result.metadata.capture_end is not None
+assert cap.result.capabilities.task_metrics.status.value == "unavailable"
+assert cap.analysis is not None
 print(f"dag:      {dfs.dag.shape}")
 print(f"combined: {dfs.combined.shape}")
+print(
+    f"capture:  {cap.result.metadata.capture_id} ({cap.result.metadata.status.value})"
+)
 
 # COMMAND ----------
 
 print("node types seen:")
 display(
-    dfs.dag.select("node_id", "node_type", "node_name", "node_duration_minutes").sort("node_id")
+    dfs.dag.select("node_id", "node_type", "node_name", "node_duration_minutes").sort(
+        "node_id"
+    )
 )
 
 # COMMAND ----------
@@ -79,7 +89,9 @@
 
 print("all dag nodes (node_type repr + raw_name):")
 for row in dfs.dag.sort("node_id").to_dicts():
-    detail = (json.loads(row["details"]).get("detail") or {}) if row.get("details") else {}
+    detail = (
+        (json.loads(row["details"]).get("detail") or {}) if row.get("details") else {}
+    )
     print(
         f"  [{row['node_id']:3d}] type={row['node_type']!r:30s}  raw_name={detail.get('raw_name', '')!r}"
     )
@@ -117,3 +129,134 @@
 # COMMAND ----------
 
 displayHTML(plot_dag(dfs))
+
+# COMMAND ----------
+
+# Brief 01 contract checks against real serverless execution. Only task-local
+# temporary artifacts are written; existing source tables are read-only.
+import tempfile
+from pathlib import Path
+
+from sparkparse import history
+from sparkparse.capture import capture
+from sparkparse.models import CaptureResult
+
+
+def assert_round_trip(result):
+    restored = CaptureResult.model_validate_json(result.model_dump_json())
+    assert restored.dag.schema == result.dag.schema
+    assert restored.combined.schema == result.combined.schema
+    assert restored.dag.equals(result.dag)
+    assert restored.combined.equals(result.combined)
+
+
+assert cap.result.dag.height > 0, "Must test populated results, not an empty success"
+assert_round_trip(cap.result)
+assert cap.analysis["metadata"] == cap.result.metadata.model_dump(mode="json")
+assert {"capture_id", "plan_version", "source_execution_id"} <= set(
+    cap.result.dag.columns
+)
+
+client = spark._client
+original_metrics = client._build_metrics
+original_table = client.to_table
+
+
+def assert_hooks_restored():
+    assert client._build_metrics == original_metrics
+    assert client.to_table == original_table
+
+
+def workload():
+    # Reuse the validated Photon join shape, with a small returned result.
+    return (
+        spark.read.table("sparkparse_demo.test.left_trips")
+        .join(spark.read.table("sparkparse_demo.test.right_trips"), on="pickup_zip")
+        .limit(5)
+        .collect()
+    )
+
+
+with tempfile.TemporaryDirectory(prefix="sparkparse-brief01-") as artifacts:
+    rules = Path(artifacts) / "alerts.toml"
+    rules.write_text("""[[alerts]]
+name = "unavailable-task-metric"
+log_name = "brief01"
+metric = "bytes_read"
+condition = "threshold"
+threshold = -1
+on_trigger = "raise"
+""")
+    history_path = str(Path(artifacts) / "history")
+    with SparkparseCapture(
+        "analyze",
+        spark=spark,
+        backend="connect",
+        log_name="brief01",
+        history_path=history_path,
+        alert_config=str(rules),
+    ) as recorded:
+        assert workload()
+    assert_hooks_restored()
+    assert recorded.result is not None
+    assert recorded.last_record is not None
+    assert recorded.result.dag.height > 0
+    assert recorded.last_record.run_id == recorded.result.metadata.capture_id
+    assert recorded.last_record.bytes_read is None
+    assert recorded.last_record.n_tasks is None
+    assert recorded.last_record.duration_s is None
+    assert recorded.triggered_alerts == []
+    stored = history.read(history_path)
+    assert stored.height == 1
+    assert stored["bytes_read"][0] is None
+    assert_round_trip(recorded.result)
+
+    with SparkparseCapture("viz", spark=spark, backend="connect") as visual:
+        assert workload()
+    assert_hooks_restored()
+    assert visual.result is not None
+    assert visual.result.dag.height > 0
+    assert visual.report and "task_metrics" in visual.report
+    assert "No plan captured" not in visual.report
+
+    with SparkparseCapture("get", spark=spark, backend="connect") as empty:
+        pass
+    assert_hooks_restored()
+    assert empty.result is not None
+    assert empty.result.dag.is_empty()
+    assert_round_trip(empty.result)
+
+    class WorkloadFailure(Exception):
+        pass
+
+    failed = SparkparseCapture("get", spark=spark, backend="connect")
+    try:
+        with failed:
+            assert workload()
+            raise WorkloadFailure("expected verification failure")
+    except WorkloadFailure:
+        pass
+    else:
+        raise AssertionError("Workload exception was suppressed")
+    assert_hooks_restored()
+    assert failed.result is not None
+    assert failed.result.metadata.status.value == "failed"
+    assert failed.result.dag.height > 0
+
+    @capture(action="get", spark=spark, backend="connect")
+    def decorated():
+        return workload()
+
+    _, first = decorated()
+    _, second = decorated()
+    assert first.result is not None
+    assert second.result is not None
+    assert first.result.metadata.capture_id != second.result.metadata.capture_id
+    assert_hooks_restored()
+    assert workload(), "Borrowed session must remain usable after capture"
+
+print(
+    "BRIEF01_CONTRACT_VERIFIED: populated/empty serialization, analysis metadata, "
+    "history nulls, alert suppression, HTML report, exception recovery, "
+    "fresh decorator state, session preservation, and hook restoration"
+)

notebooks/sparkparse_demo.py

@@ -172,7 +172,7 @@
 # MAGIC ## Live capture: NYC taxi data
 # MAGIC
 # MAGIC `SparkparseCapture` auto-detects the runtime:
-# MAGIC - **Classic cluster**: restarts the SparkSession to enable event logging (resets in-memory DataFrames). Use `cap.spark` inside the block.
+# MAGIC - **Classic cluster**: borrows the existing SparkSession and requires event logging to be enabled before capture. Use `cap.spark` inside the block.
 # MAGIC - **Serverless**: captures plan-level metrics via Spark Connect without restarting the session. `cap.spark` is the same session.
 
 # COMMAND ----------
@@ -191,7 +191,7 @@
     )
     display(result.limit(20))
 
-dfs_taxi = cap._parsed_logs
+dfs_taxi = cap.dfs
 assert dfs_taxi is not None
 print(f"dag:      {dfs_taxi.dag.shape}")
 print(f"combined: {dfs_taxi.combined.shape}")
@@ -260,7 +260,7 @@
     display(exploded.limit(20))
     print(f"exploded row count: {exploded.count():,}")
 
-dfs_explosion = cap._parsed_logs
+dfs_explosion = cap.dfs
 assert dfs_explosion is not None
 print(f"dag:      {dfs_explosion.dag.shape}")
 print(f"combined: {dfs_explosion.combined.shape}")

plans/improvements-2026-09/01-capture-and-capabilities.md

@@ -1,8 +1,22 @@
 # 01 — Safe capture and an explicit capability contract
 
-Status: proposed. Priority: P0. Primary files: `capture.py`, `models.py`,
+Status: implemented; offline and Databricks Free serverless smoke-tested. Priority: P0. Primary files: `capture.py`, `models.py`,
 `connect.py`, `analyze.py`, `history.py`, and public exports.
 
+Review corrections: owned logs are parsed after session shutdown; borrowed logs
+are selected by application ID, not filename ordering. Explicit backend selection,
+session-free file ingestion, capture-error policy, schema-preserving JSON, and
+in-memory HTML reports are implemented. Analysis/history consume coverage and
+retain unavailable totals as null. Coverage remains conservative when completeness
+cannot be established. These corrections have local regression coverage and passed
+Databricks Free serverless run `566565689796128` on 2026-09-12. The expanded notebook
+verified populated/empty serialization, analysis metadata, history nulls and alert
+suppression, HTML reports, exception recovery, fresh decorator state, session
+preservation, and hook restoration. The first remote run exposed timezone inference
+failure in history reads with newer Polars; explicit UTC parsing fixed it, with a
+local offset-normalization regression test. Classic shutdown/log-flushing behavior
+is covered locally, not by this Connect-only remote run.
+
 ## Evidence and problem
 
 `SparkparseCapture.__enter__` stops the supplied classic session and recreates it.
@@ -22,9 +36,10 @@ defined common contract plus optional tables/fields.
 ## Design
 
 Introduce a versioned capture result containing `dag`, `combined`, metadata,
-capabilities, and diagnostics. Retain `ParsedLogDataFrames` compatibility through a
-documented adapter and retain `_parsed_logs` temporarily as an alias. Expose public
-`cap.dfs`, `cap.analysis`, `cap.result`, `cap.last_record`, and `cap.triggered_alerts`.
+capabilities, and diagnostics. `ParsedLogDataFrames` remains an internal parser/table
+container; the capture API exposes the new result directly through
+`cap.result`, alongside `cap.dfs`, `cap.analysis`, `cap.last_record`, and
+`cap.triggered_alerts`. No legacy capture aliases are maintained.
 Use one source-neutral finalizer for get/analyze/report, history, and alerts.
 
 Metadata should contain capture/run ID, source application/session ID if observed,
@@ -64,9 +79,9 @@ nested capture deliberately, and create fresh state for each decorator invocatio
 
 ## Increments and acceptance
 
-1. Add metadata/capability models and adapters. Existing constructors still work;
-   legacy data has unknown coverage rather than assumed completeness. Contract tests
-   cover empty, classic, and Connect-shaped results and serialization round trips.
+1. Add metadata/capability models. Legacy capture aliases are intentionally removed;
+   raw parser tables have unknown coverage rather than assumed completeness. Contract
+   tests cover empty, classic, and Connect-shaped results and serialization round trips.
 2. Refactor the common finalizer. Mock both backends and verify get/analyze/history/
    alerts run equivalently without requiring a log directory. Invalid actions fail
    at construction. Missing coverage produces diagnostics, never invented totals.

plans/improvements-2026-09/README.md

@@ -1,6 +1,7 @@
 # Project review and implementation queue
 
-Reviewed 2026-09-12 at commit `129318c`. Status: proposals, not implemented.
+Reviewed 2026-09-12 at commit `129318c`. Brief 01 is implemented and serverless-smoke
+validated; the remaining briefs are proposals.
 This queue supersedes the pending-work descriptions in `IMPLEMENTATION.md` and
 `plans/SPARK_CONNECT_CAPTURE.md`; retain those documents as historical context.
 
@@ -20,7 +21,7 @@ access mode. Spark Connect is a transport, not proof that compute is serverless.
 
 | Priority | Brief | Outcome | Relative size / dependency |
 |---|---|---|---|
-| P0 | [01 — Capture and capabilities](01-capture-and-capabilities.md) | Safe session lifecycle, common finalization, explicit missing data | Large; foundation |
+| P0 | [01 — Capture and capabilities](01-capture-and-capabilities.md) | Safe session lifecycle, common finalization, explicit missing data | Implemented + smoke-tested |
 | P0 | [02 — Connect correctness](02-connect-correctness.md) | Reliable query attribution and tolerant operator handling | Medium–large; 01 contract |
 | P0 | [03 — Analysis correctness and depth](03-analysis.md) | Accurate metrics and evidence-based findings | Large; 01–02 |
 | P1 | [04 — Event-log robustness](04-event-logs.md) | Non-AQE, partial, rolled, retried, and larger workloads | Large; 01 identities |

sparkparse/__init__.py

@@ -1,3 +1,23 @@
 from sparkparse.capture import SparkparseCapture, capture, capture_context
+from sparkparse.models import (
+    CapabilityStatus,
+    CaptureCapabilities,
+    CaptureCapability,
+    CaptureDiagnostic,
+    CaptureMetadata,
+    CaptureResult,
+    CaptureStatus,
+)
 
-__all__ = ["SparkparseCapture", "capture", "capture_context"]
+__all__ = [
+    "CaptureCapabilities",
+    "CaptureCapability",
+    "CaptureDiagnostic",
+    "CaptureMetadata",
+    "CaptureResult",
+    "CaptureStatus",
+    "CapabilityStatus",
+    "SparkparseCapture",
+    "capture",
+    "capture_context",
+]

sparkparse/alerts.py

@@ -161,7 +161,15 @@ def check_alerts(
             )
             continue
 
-        current = float(getattr(record, alert.metric))
+        current_value = getattr(record, alert.metric)
+        if current_value is None:
+            logger.warning(
+                "Alert '%s' references unavailable metric '%s', skipping",
+                alert.name,
+                alert.metric,
+            )
+            continue
+        current = float(current_value)
         baseline: float | None = None
 
         if alert.condition == "threshold":