fix: show runtime-detected assets in a run's Assets tab (#10738)

* fix: show runtime-detected assets in a run's Assets tab

* fix: address review nits on run assets tab

* fix: cap the run assets list and report when it is cut

* fix: cap run assets by asset, not by row

backend/.sqlx/query-0df1c23ac429b5807d42e1da5ee2e2176a8a774f107bece5804cc7b221365c89.json

@@ -0,0 +1,58 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "WITH RECURSIVE job_tree AS (\n            SELECT id, tag FROM v2_job WHERE id = $2 AND workspace_id = $1\n            UNION\n            SELECT j.id, j.tag FROM v2_job j JOIN job_tree t ON j.parent_job = t.id\n             WHERE j.workspace_id = $1\n        )\n        SELECT\n            a.path,\n            a.kind AS \"kind!: windmill_common::assets::AssetKind\",\n            -- Several jobs of the tree touch one asset, each recording its own\n            -- access. A job that recorded none contributes nothing rather than\n            -- erasing a sibling's, so an all-null group is the only unknown one.\n            -- Grouping here, not in Rust, is what makes LIMIT count assets: the\n            -- retention keeps up to ten job rows per asset.\n            COALESCE(bool_or(a.usage_access_type IN ('r', 'rw')), false) AS \"any_read!\",\n            COALESCE(bool_or(a.usage_access_type IN ('w', 'rw')), false) AS \"any_write!\"\n        FROM asset a JOIN job_tree t ON a.usage_path = t.id::text\n        WHERE a.workspace_id = $1 AND a.usage_kind = 'job'\n          AND ($3::text[] IS NULL OR t.tag = ANY($3))\n        GROUP BY a.path, a.kind\n        ORDER BY a.path, a.kind\n        LIMIT $4",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "path",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 1,
+        "name": "kind!: windmill_common::assets::AssetKind",
+        "type_info": {
+          "Custom": {
+            "name": "asset_kind",
+            "kind": {
+              "Enum": [
+                "s3object",
+                "resource",
+                "variable",
+                "ducklake",
+                "datatable",
+                "volume",
+                "dbt"
+              ]
+            }
+          }
+        }
+      },
+      {
+        "ordinal": 2,
+        "name": "any_read!",
+        "type_info": "Bool"
+      },
+      {
+        "ordinal": 3,
+        "name": "any_write!",
+        "type_info": "Bool"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Uuid",
+        "TextArray",
+        "Int8"
+      ]
+    },
+    "nullable": [
+      false,
+      false,
+      null,
+      null
+    ]
+  },
+  "hash": "0df1c23ac429b5807d42e1da5ee2e2176a8a774f107bece5804cc7b221365c89"
+}

backend/windmill-api-integration-tests/tests/assets.rs

@@ -0,0 +1,248 @@
+use serde_json::{json, Value};
+use sqlx::{Pool, Postgres};
+use uuid::Uuid;
+use windmill_test_utils::*;
+
+fn client() -> reqwest::Client {
+    reqwest::Client::new()
+}
+
+fn bearer(builder: reqwest::RequestBuilder, token: &str) -> reqwest::RequestBuilder {
+    builder.header("Authorization", format!("Bearer {token}"))
+}
+
+async fn insert_job(db: &Pool<Postgres>, parent: Option<Uuid>, tag: &str) -> anyhow::Result<Uuid> {
+    let id = Uuid::new_v4();
+    sqlx::query(
+        "INSERT INTO v2_job (id, workspace_id, tag, created_by, permissioned_as, \
+         permissioned_as_email, kind, parent_job, same_worker, visible_to_owner) \
+         VALUES ($1, 'test-workspace', $3, 'test-user', 'u/test-user', \
+         'test@windmill.dev', 'script', $2, false, true)",
+    )
+    .bind(id)
+    .bind(parent)
+    .bind(tag)
+    .execute(db)
+    .await?;
+    Ok(id)
+}
+
+/// `access_type` is `None` for a detection that could not tell read from write
+/// — how a resource passed in a job's arguments is recorded.
+async fn insert_job_asset(
+    db: &Pool<Postgres>,
+    job: Uuid,
+    path: &str,
+    access_type: Option<&str>,
+) -> anyhow::Result<()> {
+    sqlx::query(
+        "INSERT INTO asset (workspace_id, path, kind, usage_access_type, usage_path, usage_kind) \
+         VALUES ('test-workspace', $1, 's3object', $2::text::asset_access_type, $3, 'job')",
+    )
+    .bind(path)
+    .bind(access_type)
+    .bind(job.to_string())
+    .execute(db)
+    .await?;
+    Ok(())
+}
+
+/// A run reports what its whole job tree touched: runtime detection records
+/// against the job that did the operation, which for a flow step or a
+/// workflow-as-code task is never the job the user opened.
+#[sqlx::test(migrations = "../migrations", fixtures("base"))]
+async fn test_list_run_assets_covers_child_jobs(db: Pool<Postgres>) -> anyhow::Result<()> {
+    initialize_tracing().await;
+    let server = ApiServer::start(db.clone()).await?;
+    let port = server.addr.port();
+    let ws = format!("http://localhost:{port}/api/w/test-workspace");
+
+    let parent = insert_job(&db, None, "other").await?;
+    let child = insert_job(&db, Some(parent), "other").await?;
+    let unrelated = insert_job(&db, None, "other").await?;
+
+    insert_job_asset(&db, parent, "/data/shared.json", Some("r")).await?;
+    insert_job_asset(&db, child, "/data/shared.json", Some("w")).await?;
+    insert_job_asset(&db, child, "/data/child_only.json", Some("w")).await?;
+    insert_job_asset(&db, parent, "/data/from_args.json", None).await?;
+    insert_job_asset(&db, child, "/data/from_args.json", Some("w")).await?;
+    insert_job_asset(&db, unrelated, "/data/unrelated.json", Some("w")).await?;
+
+    let resp = bearer(
+        client().get(format!("{ws}/jobs/run_assets/{parent}")),
+        "SECRET_TOKEN",
+    )
+    .send()
+    .await?;
+    assert_eq!(resp.status().as_u16(), 200);
+    let body: Value = resp.json().await?;
+    assert_eq!(body["truncated"], json!(false));
+    assert_eq!(
+        body["assets"],
+        json!([
+            { "path": "/data/child_only.json", "kind": "s3object", "access_type": "w" },
+            // The parent recorded no access type for this one; that must not erase
+            // the child's.
+            { "path": "/data/from_args.json", "kind": "s3object", "access_type": "w" },
+            { "path": "/data/shared.json", "kind": "s3object", "access_type": "rw" },
+        ]),
+        "parent should report its own and its child's assets, with access types merged"
+    );
+
+    let resp = bearer(
+        client().get(format!("{ws}/jobs/run_assets/{child}")),
+        "SECRET_TOKEN",
+    )
+    .send()
+    .await?;
+    assert_eq!(resp.status().as_u16(), 200);
+    let body: Value = resp.json().await?;
+    assert_eq!(
+        body["assets"],
+        json!([
+            { "path": "/data/child_only.json", "kind": "s3object", "access_type": "w" },
+            { "path": "/data/from_args.json", "kind": "s3object", "access_type": "w" },
+            { "path": "/data/shared.json", "kind": "s3object", "access_type": "w" },
+        ]),
+        "a child should report only what it touched itself"
+    );
+
+    // `asset` has no RLS of its own, so the job read gate is the only thing
+    // standing between another member and these paths.
+    let resp = bearer(
+        client().get(format!("{ws}/jobs/run_assets/{parent}")),
+        "SECRET_TOKEN_2",
+    )
+    .send()
+    .await?;
+    assert_eq!(
+        resp.status().as_u16(),
+        403,
+        "a member who cannot read the run must not read its assets"
+    );
+
+    // ...and a share link is what lets that same member in. The tree is walked
+    // outside the caller's RLS precisely so this works, since the token grants
+    // access their own permissions do not.
+    let view_token: String = bearer(
+        client().get(format!("{ws}/jobs/job_view_token/{parent}")),
+        "SECRET_TOKEN",
+    )
+    .send()
+    .await?
+    .text()
+    .await?;
+    let resp = bearer(
+        client().get(format!("{ws}/jobs/run_assets/{parent}")),
+        "SECRET_TOKEN_2",
+    )
+    .header("X-View-Token", view_token)
+    .send()
+    .await?;
+    assert_eq!(resp.status().as_u16(), 200);
+    let body: Value = resp.json().await?;
+    assert_eq!(
+        body["assets"].as_array().map(|a| a.len()),
+        Some(3),
+        "a share-link viewer should see the whole tree's assets"
+    );
+
+    Ok(())
+}
+
+/// The read gate only checks the root job's tag, so the walk has to keep a
+/// tag-scoped token out of descendants outside its scope — without hiding the
+/// ones below them, which the token could have asked for directly.
+#[sqlx::test(migrations = "../migrations", fixtures("base"))]
+async fn test_list_run_assets_scopes_descendants_by_tag(db: Pool<Postgres>) -> anyhow::Result<()> {
+    initialize_tracing().await;
+    let server = ApiServer::start(db.clone()).await?;
+    let port = server.addr.port();
+    let ws = format!("http://localhost:{port}/api/w/test-workspace");
+
+    sqlx::query(
+        "INSERT INTO token (token_hash, token_prefix, token, email, label, super_admin, scopes) \
+         VALUES (encode(sha256('TAG_TOKEN'::bytea), 'hex'), 'TAG_TOK', 'TAG_TOKEN', \
+         'test@windmill.dev', 'tag-only', true, ARRAY['if_jobs:filter_tags:other'])",
+    )
+    .execute(&db)
+    .await?;
+
+    let parent = insert_job(&db, None, "other").await?;
+    let in_scope = insert_job(&db, Some(parent), "other").await?;
+    let out_of_scope = insert_job(&db, Some(parent), "deno").await?;
+    let below_out_of_scope = insert_job(&db, Some(out_of_scope), "other").await?;
+    insert_job_asset(&db, in_scope, "/data/in_scope.json", Some("w")).await?;
+    insert_job_asset(&db, out_of_scope, "/data/out_of_scope.json", Some("w")).await?;
+    insert_job_asset(&db, below_out_of_scope, "/data/nested.json", Some("w")).await?;
+
+    let resp = bearer(
+        client().get(format!("{ws}/jobs/run_assets/{parent}")),
+        "TAG_TOKEN",
+    )
+    .send()
+    .await?;
+    assert_eq!(resp.status().as_u16(), 200);
+    let body: Value = resp.json().await?;
+    assert_eq!(
+        body["assets"],
+        json!([
+            { "path": "/data/in_scope.json", "kind": "s3object", "access_type": "w" },
+            { "path": "/data/nested.json", "kind": "s3object", "access_type": "w" },
+        ]),
+        "a tag-scoped token must not read assets of descendants outside its tags, \
+         but must still reach in-scope jobs below them"
+    );
+
+    Ok(())
+}
+
+/// A fan-out run can touch more assets than one response should carry, and the
+/// cut must be reported rather than served as if it were the whole list.
+#[sqlx::test(migrations = "../migrations", fixtures("base"))]
+async fn test_list_run_assets_caps_the_list(db: Pool<Postgres>) -> anyhow::Result<()> {
+    initialize_tracing().await;
+    let server = ApiServer::start(db.clone()).await?;
+    let port = server.addr.port();
+    let ws = format!("http://localhost:{port}/api/w/test-workspace");
+
+    // Three child jobs touching the same 1200 assets: the cap counts assets, and
+    // the retention allows ten job rows per asset, so a row-counted cap would cut
+    // this at a third of the list.
+    let parent = insert_job(&db, None, "other").await?;
+    for _ in 0..3 {
+        let child = insert_job(&db, Some(parent), "other").await?;
+        sqlx::query(
+            "INSERT INTO asset (workspace_id, path, kind, usage_access_type, usage_path, usage_kind) \
+             SELECT 'test-workspace', '/out/' || lpad(g::text, 6, '0') || '.json', 's3object', 'w', \
+             $1, 'job' FROM generate_series(1, 1200) g",
+        )
+        .bind(child.to_string())
+        .execute(&db)
+        .await?;
+    }
+
+    let resp = bearer(
+        client().get(format!("{ws}/jobs/run_assets/{parent}")),
+        "SECRET_TOKEN",
+    )
+    .send()
+    .await?;
+    assert_eq!(resp.status().as_u16(), 200);
+    let body: Value = resp.json().await?;
+    assert_eq!(body["truncated"], json!(true));
+    let assets = body["assets"].as_array().expect("assets array");
+    assert_eq!(assets.len(), 1000);
+    assert_eq!(
+        assets[0],
+        json!({ "path": "/out/000001.json", "kind": "s3object", "access_type": "w" }),
+        "the cap keeps the head of the ordered list, with its access type merged"
+    );
+    // Three jobs touched each asset, so a cap counting rows rather than assets
+    // would fill the list with repeats and stop around /out/000334.json.
+    assert_eq!(
+        assets[999]["path"], "/out/001000.json",
+        "the cap counts assets, not asset rows"
+    );
+    Ok(())
+}

backend/windmill-api/openapi.yaml

@@ -23847,6 +23847,55 @@ paths:
                 items:
                   $ref: "#/components/schemas/AssetProgress"
 
+  /w/{workspace}/jobs/run_assets/{id}:
+    get:
+      summary: List the assets a run touched at runtime
+      description: >
+        Assets detected while the run executed (SDK S3 calls, resources passed as
+        arguments), aggregated over the job and all of its child jobs so a flow or
+        workflow-as-code run reports what its steps and tasks touched. Authorized
+        through the job, the same gate as `run_progress`. Recording is asynchronous,
+        so an asset can take a few minutes after the run to appear, and only the most
+        recent runs that touched an asset keep that record. A run that fans out can
+        touch more assets than one response should carry, so the list is capped and
+        `truncated` says when it was cut.
+      operationId: listRunAssets
+      tags:
+        - job
+      parameters:
+        - $ref: "#/components/parameters/WorkspaceId"
+        - name: id
+          in: path
+          required: true
+          description: The job whose runtime assets to read
+          schema:
+            type: string
+            format: uuid
+      responses:
+        "200":
+          description: assets this run and its child jobs touched
+          content:
+            application/json:
+              schema:
+                type: object
+                required: [assets, truncated]
+                properties:
+                  truncated:
+                    type: boolean
+                    description: whether the run touched more assets than are listed
+                  assets:
+                    type: array
+                    items:
+                      type: object
+                      required: [path, kind]
+                      properties:
+                        path:
+                          type: string
+                        kind:
+                          $ref: "#/components/schemas/AssetKind"
+                        access_type:
+                          $ref: "#/components/schemas/AssetUsageAccessType"
+
   /w/{workspace}/assets/partitions_in_range:
     get:
       summary: List expected partitions of a ducklake asset in a date range with their materialization status (enterprise)

backend/windmill-api/src/jobs.rs

@@ -25,6 +25,7 @@ use std::time::Instant;
 use tokio::io::AsyncReadExt;
 use tower::ServiceBuilder;
 use url::Url;
+use windmill_common::assets::AssetUsageAccessType;
 #[cfg(all(feature = "enterprise", feature = "instance_smtp"))]
 use windmill_common::auth::is_super_admin_email;
 use windmill_common::auth::TOKEN_PREFIX_LEN;
@@ -136,6 +137,7 @@ pub fn workspaced_service() -> Router {
 
     Router::new()
         .route("/run_progress/{id}", get(get_run_progress))
+        .route("/run_assets/{id}", get(list_run_assets))
         .route("/dbt_graph/{id}", get(get_dbt_run_graph))
         .route("/dbt_resumable/{id}", get(get_dbt_resumable))
         .route(
@@ -1188,6 +1190,121 @@ async fn get_run_progress(
     Ok(Json(rows))
 }
 
+#[derive(Serialize)]
+struct RunAsset {
+    path: String,
+    kind: windmill_common::assets::AssetKind,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    access_type: Option<AssetUsageAccessType>,
+}
+
+#[derive(Serialize)]
+struct RunAssets {
+    assets: Vec<RunAsset>,
+    truncated: bool,
+}
+
+/// A fan-out run — a forloop writing one object per iteration — touches as many
+/// assets as it has steps, and the whole list would land in one response and one
+/// list in the browser. Cap it, and say so rather than serving a prefix that
+/// reads like the whole answer.
+const RUN_ASSETS_CAP: usize = 1000;
+
+/// Assets a run touched, as recorded by runtime detection, aggregated over the
+/// whole job tree: the recorder attributes an asset to the job that performed
+/// the operation, which for a flow step or a workflow-as-code task is not the
+/// job the user opened. Lives with the job routes for the same reason
+/// `run_progress` does — `asset` has no RLS, and `require_job_read_access` is
+/// what applies the view token, a scoped token's tag filter and the app-embed
+/// cutoff.
+async fn list_run_assets(
+    authed: ApiAuthed,
+    OptViewToken(view_token): OptViewToken,
+    Extension(db): Extension<DB>,
+    Extension(user_db): Extension<UserDB>,
+    Path((w_id, job_id)): Path<(String, Uuid)>,
+) -> error::JsonResult<RunAssets> {
+    let created_by = sqlx::query_scalar!(
+        "SELECT created_by FROM v2_job WHERE id = $1 AND workspace_id = $2",
+        job_id,
+        &w_id
+    )
+    .fetch_optional(&db)
+    .await?;
+    let Some(created_by) = created_by else {
+        return Ok(Json(RunAssets { assets: vec![], truncated: false }));
+    };
+    require_job_read_access(
+        &db,
+        &user_db,
+        &authed,
+        &w_id,
+        &job_id,
+        &created_by,
+        view_token.as_deref(),
+    )
+    .await?;
+
+    // Walked on `db`, like the flow tree is: the gate above is what authorizes the
+    // run, and it grants access the caller's own RLS does not have — a view token,
+    // or a job the caller launched that runs as someone else. Re-filtering the tree
+    // through `user_db` would drop exactly those, and hand a share-link viewer the
+    // empty tab this endpoint exists to fix. The tag scope is the one restriction
+    // that must still hold per job, since the gate only checked the root. It gates
+    // which jobs' assets are read, not which are walked through: the gate admits a
+    // job on its own tag, so an out-of-scope job in the middle of the tree must not
+    // hide a descendant the caller could have asked for directly.
+    let scope_tags = get_scope_tags(&authed).map(|v| v.iter().map(|s| s.to_string()).collect_vec());
+    let rows = sqlx::query!(
+        r#"WITH RECURSIVE job_tree AS (
+            SELECT id, tag FROM v2_job WHERE id = $2 AND workspace_id = $1
+            UNION
+            SELECT j.id, j.tag FROM v2_job j JOIN job_tree t ON j.parent_job = t.id
+             WHERE j.workspace_id = $1
+        )
+        SELECT
+            a.path,
+            a.kind AS "kind!: windmill_common::assets::AssetKind",
+            -- Several jobs of the tree touch one asset, each recording its own
+            -- access. A job that recorded none contributes nothing rather than
+            -- erasing a sibling's, so an all-null group is the only unknown one.
+            -- Grouping here, not in Rust, is what makes LIMIT count assets: the
+            -- retention keeps up to ten job rows per asset.
+            COALESCE(bool_or(a.usage_access_type IN ('r', 'rw')), false) AS "any_read!",
+            COALESCE(bool_or(a.usage_access_type IN ('w', 'rw')), false) AS "any_write!"
+        FROM asset a JOIN job_tree t ON a.usage_path = t.id::text
+        WHERE a.workspace_id = $1 AND a.usage_kind = 'job'
+          AND ($3::text[] IS NULL OR t.tag = ANY($3))
+        GROUP BY a.path, a.kind
+        ORDER BY a.path, a.kind
+        LIMIT $4"#,
+        w_id,
+        job_id,
+        scope_tags.as_deref(),
+        // One asset past the cap is how the response learns it was cut.
+        RUN_ASSETS_CAP as i64 + 1
+    )
+    .fetch_all(&db)
+    .await?;
+
+    let truncated = rows.len() > RUN_ASSETS_CAP;
+    let assets = rows
+        .into_iter()
+        .take(RUN_ASSETS_CAP)
+        .map(|row| RunAsset {
+            path: row.path,
+            kind: row.kind,
+            access_type: match (row.any_read, row.any_write) {
+                (true, true) => Some(AssetUsageAccessType::RW),
+                (true, false) => Some(AssetUsageAccessType::R),
+                (false, true) => Some(AssetUsageAccessType::W),
+                (false, false) => None,
+            },
+        })
+        .collect();
+    Ok(Json(RunAssets { assets, truncated }))
+}
+
 async fn get_flow_job_debug_info(
     OptViewToken(view_token): OptViewToken,
     OptAuthed(opt_authed): OptAuthed,
@@ -1654,20 +1771,21 @@ async fn require_job_within_run_scope(
     .await?;
 
     let runs_app = build_scope_path_predicate(authed, "apps", "run");
-    let in_scope = chain.iter().any(|job| {
-        match (job.runnable_path.as_deref(), job.scope_kind.as_deref()) {
-            (Some(runnable_path), Some(kind))
-                if windmill_api_auth::scopes::run_confinement_admits(
-                    &confinement,
-                    kind,
-                    runnable_path,
-                ) =>
-            {
-                true
-            }
-            _ => job.launched_by_app.as_deref().is_some_and(&runs_app),
-        }
-    });
+    let in_scope =
+        chain.iter().any(
+            |job| match (job.runnable_path.as_deref(), job.scope_kind.as_deref()) {
+                (Some(runnable_path), Some(kind))
+                    if windmill_api_auth::scopes::run_confinement_admits(
+                        &confinement,
+                        kind,
+                        runnable_path,
+                    ) =>
+                {
+                    true
+                }
+                _ => job.launched_by_app.as_deref().is_some_and(&runs_app),
+            },
+        );
 
     if in_scope {
         Ok(())

frontend/src/lib/components/assets/JobAssetsViewer.svelte

@@ -1,13 +1,15 @@
 <script lang="ts">
-	import { ResourceService, ScriptService, type Job } from '$lib/gen'
+	import { JobService, ResourceService, ScriptService, type Job } from '$lib/gen'
 	import { inferAssets } from '$lib/infer'
 	import { workspaceStore } from '$lib/stores'
 	import { usePromise } from '$lib/svelte5Utils.svelte'
 	import { pruneNullishArray, uniqueBy } from '$lib/utils'
+	import { Skeleton } from '../common'
 	import ResourceEditorDrawer from '../ResourceEditorDrawer.svelte'
 	import S3FilePicker from '../S3FilePicker.svelte'
 	import AssetButtons from './AssetButtons.svelte'
 	import {
+		formatAssetAccessType,
 		formatAssetKind,
 		getFlowModuleAssets,
 		parseInputArgsAssets,
@@ -19,7 +21,52 @@
 	}
 	let { job }: Props = $props()
 
-	async function extractAssets(job: Job): Promise<AssetWithAccessType[]> {
+	const assetKey = (a: AssetWithAccessType) => a.kind + a.path
+
+	async function extractAssets(
+		job: Job
+	): Promise<{ assets: AssetWithAccessType[]; truncated: boolean }> {
+		const [runtime, staticAssets] = await Promise.all([
+			fetchRuntimeAssets(job),
+			inferStaticAssets(job)
+		])
+		// Runtime assets win: they are what the run actually did. Their access type
+		// can still be unknown — a resource passed in the arguments is recorded
+		// without one — so fall back to what the parser inferred for the same asset.
+		// One asset can appear twice statically, parsed from the code and again from
+		// the arguments; only the parsed one carries an access type.
+		const staticByKey = new Map<string, AssetWithAccessType>()
+		for (const a of staticAssets) {
+			const prev = staticByKey.get(assetKey(a))
+			if (!prev || (!prev.access_type && a.access_type)) staticByKey.set(assetKey(a), a)
+		}
+		const merged = runtime.assets.map((a) => ({
+			...a,
+			access_type: a.access_type ?? staticByKey.get(assetKey(a))?.access_type
+		}))
+		return {
+			assets: uniqueBy([...merged, ...staticAssets], assetKey),
+			truncated: runtime.truncated
+		}
+	}
+
+	// What the run touched, as recorded by runtime detection. Static inference
+	// misses these whenever the path is only known at runtime, and it never sees
+	// what a flow step or a workflow-as-code task did on the parent's behalf.
+	async function fetchRuntimeAssets(
+		job: Job
+	): Promise<{ assets: AssetWithAccessType[]; truncated: boolean }> {
+		if (!$workspaceStore) return { assets: [], truncated: false }
+		return await JobService.listRunAssets({
+			workspace: $workspaceStore,
+			id: job.id
+		}).catch((err) => {
+			console.error("Couldn't fetch runtime assets of job", job.id, err)
+			return { assets: [], truncated: false }
+		})
+	}
+
+	async function inferStaticAssets(job: Job): Promise<AssetWithAccessType[]> {
 		if (job.job_kind === 'flow') {
 			const additionalAssetsMap = {
 				// TODO : Transitive assets
@@ -59,7 +106,7 @@
 
 	let resourceDataCache: Record<string, string | undefined> = $state({})
 	$effect(() => {
-		for (const asset of assets.value ?? []) {
+		for (const asset of assets.value?.assets ?? []) {
 			if (asset.kind == 'resource') {
 				let truncatedPath = asset.path.split('?table=')[0]
 				if (truncatedPath in resourceDataCache) continue
@@ -75,10 +122,12 @@
 	let resourceEditorDrawer: ResourceEditorDrawer | undefined = $state()
 </script>
 
-{#if assets.value && assets.value.length > 0}
+{#if assets.status === 'idle' || assets.status === 'loading'}
+	<Skeleton layout={[[3], 0.5, [3]]} class="w-full" />
+{:else if assets.value && assets.value.assets.length > 0}
 	<ul class="flex flex-col divide-y mt-1">
-		{#each assets.value ?? [] as asset}
-			<li class="flex justify-between py-3 leading-4 text-sm pl-4">
+		{#each assets.value.assets as asset}
+			<li class="flex justify-between items-center gap-2 py-3 leading-4 text-sm pl-4">
 				<div class="flex flex-col flex-1 truncate">
 					{asset.path}
 					<span class="text-2xs text-primary">
@@ -90,17 +139,26 @@
 						})}
 					</span>
 				</div>
-				<AssetButtons
-					{asset}
-					{resourceDataCache}
-					{resourceEditorDrawer}
-					{s3FilePicker}
-				/>
+				{#if asset.access_type}
+					<span class="text-xs text-secondary">{formatAssetAccessType(asset.access_type)}</span>
+				{/if}
+				<AssetButtons {asset} {resourceDataCache} {resourceEditorDrawer} {s3FilePicker} />
 			</li>
 		{/each}
 	</ul>
+	{#if assets.value.truncated}
+		<div class="text-2xs text-secondary mt-2 pl-4">
+			This run touched more assets than are listed here.
+		</div>
+	{/if}
 {:else}
-	<div class="text-sm text-primary">No assets found</div>
+	<div class="flex flex-col gap-1">
+		<span class="text-sm text-primary">No assets found</span>
+		<span class="text-2xs text-secondary">
+			Assets detected while a run executes are recorded asynchronously, and only the most recent
+			runs that touched an asset keep that record.
+		</span>
+	</div>
 {/if}
 
 <S3FilePicker bind:this={s3FilePicker} readOnlyMode />