UI: Add a column in asset store display table linking to the TI that wrote it

airflow-core/src/airflow/ui/public/i18n/locales/en/assets.json

@@ -13,6 +13,9 @@
     "deleteWarning": "The asset will lose this persisted store entry.",
     "edit": "Edit Asset Store",
     "emptyState": "Asset store stores values scoped to an asset identity, shared across all Dag runs. Workers can write asset store via the Task SDK.",
+    "lastUpdatedBy": "Last Updated By",
+    "lastUpdatedByApi": "API",
+    "lastUpdatedByWatcher": "Watcher",
     "title": "Asset Store"
   },
   "consumingDags": "Consuming Dags",

airflow-core/src/airflow/ui/src/pages/Asset/AssetStore/AssetStore.tsx

@@ -16,24 +16,36 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-import { Flex, Text } from "@chakra-ui/react";
+import { Flex, Link, Text } from "@chakra-ui/react";
 import type { ColumnDef } from "@tanstack/react-table";
 import { useTranslation } from "react-i18next";
-import { useParams } from "react-router-dom";
+import { Link as RouterLink, useParams } from "react-router-dom";
 
 import { useAssetStoreServiceListAssetStore } from "openapi/queries";
-import type { AssetStoreResponse } from "openapi/requests";
+import type { AssetStoreLastUpdatedBy, AssetStoreResponse } from "openapi/requests";
 import { DataTable } from "src/components/DataTable";
 import { useTableURLState } from "src/components/DataTable/useTableUrlState";
 import { ErrorAlert } from "src/components/ErrorAlert";
 import { StoreValueCell } from "src/components/StoreValueCell";
 import Time from "src/components/Time";
+import { getTaskInstanceLink } from "src/utils/links";
 
 import { AddAssetStoreButton } from "./AddAssetStoreButton";
 import { ClearAllAssetStoreButton } from "./ClearAllAssetStoreButton";
 import { DeleteAssetStoreButton } from "./DeleteAssetStoreButton";
 import { EditAssetStoreButton } from "./EditAssetStoreButton";
 
+type TaskWriter = { dag_id: string; run_id: string; task_id: string } & AssetStoreLastUpdatedBy;
+
+const isTaskWriter = (writer: AssetStoreLastUpdatedBy): writer is TaskWriter =>
+  writer.kind === "task" &&
+  writer.dag_id !== null &&
+  writer.dag_id !== undefined &&
+  writer.run_id !== null &&
+  writer.run_id !== undefined &&
+  writer.task_id !== null &&
+  writer.task_id !== undefined;
+
 type ColumnsProps = {
   readonly assetId: number;
   readonly translate: (key: string) => string;
@@ -56,6 +68,45 @@ const getColumns = ({ assetId, translate }: ColumnsProps): Array<ColumnDef<Asset
     cell: ({ row: { original } }) => <Time datetime={original.updated_at} />,
     header: translate("common:table.updatedAt"),
   },
+  {
+    accessorKey: "last_updated_by",
+    cell: ({ row: { original } }) => {
+      const writer = original.last_updated_by;
+
+      if (!writer) {
+        return <Text color="fg.muted">—</Text>;
+      }
+      if (isTaskWriter(writer)) {
+        const path = getTaskInstanceLink({
+          dagId: writer.dag_id,
+          dagRunId: writer.run_id,
+          mapIndex: writer.map_index ?? undefined,
+          taskId: writer.task_id,
+        });
+
+        return (
+          <Flex direction="column">
+            <Link asChild color="fg.info">
+              <RouterLink to={path}>{writer.task_id}</RouterLink>
+            </Link>
+            <Text color="fg.muted" fontSize="xs">
+              {writer.dag_id}
+            </Text>
+          </Flex>
+        );
+      }
+
+      return (
+        <Text>
+          {writer.kind === "api"
+            ? translate("assets:assetStore.lastUpdatedByApi")
+            : translate("assets:assetStore.lastUpdatedByWatcher")}
+        </Text>
+      );
+    },
+    enableSorting: false,
+    header: translate("assets:assetStore.lastUpdatedBy"),
+  },
   {
     accessorKey: "actions",
     cell: ({ row: { original } }) => (