H-3848, H-6154: Handle larger Flow payloads, historical flight data Flows #8356

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged

CiaranMn merged 9 commits into main from cm/historical-flights-and-large-flow-improvements

Feb 5, 2026

.cursor/rules/typescript-coding-guidelines.mdc

-Original file line number
+Diff line change
@@ Expand Up / @@ -26,7 +26,7 @@ alwaysApply: false @@
     # Workspaces
     - The project uses yarn workspaces.
-    - If you want to install a dependency, you need to do it in the relevant workspace. e.g. `hash/apps/hash-frontend`.
+    - If you want to install a dependency, you need to do it in the relevant workspace. e.g. `apps/hash-frontend`.
     - The project
     # Frontend
@@ Expand Down @@

apps/hash-ai-worker-ts/src/activities/flow-activities/answer-question-action.ts

-Original file line number
+Diff line change
@@ -1,5 +1,9 @@
     import { extractEntityUuidFromEntityId } from "@blockprotocol/type-system";
     import type { AiFlowActionActivity } from "@local/hash-backend-utils/flows";
+    import {
+      getStorageProvider,
+      resolvePayloadValue,
+    } from "@local/hash-backend-utils/flows/payload-storage";
     import { getSimpleGraph } from "@local/hash-backend-utils/simplified-graph";
     import { queryEntitySubgraph } from "@local/hash-graph-sdk/entity";
     import type { AiActionStepOutput } from "@local/hash-isomorphic-utils/flows/action-definitions";
@@ Expand Down Expand Up / @@ -401,7 +405,7 @@ export const answerQuestionAction: AiFlowActionActivity< @@
     > = async ({ inputs }) => {
       const {
         context,
-        entities: inputEntities,
+        entities: entitiesInput,
         question,
       } = getSimplifiedAiFlowActionInputs({
         inputs,
@@ Expand All / @@ -410,6 +414,15 @@ export const answerQuestionAction: AiFlowActionActivity< @@
       const { userAuthentication } = await getFlowContext();
+      // Resolve the stored ref to get the array of PersistedEntitiesMetadata
+      const inputEntities = entitiesInput
+        ? await resolvePayloadValue(
+            getStorageProvider(),
+            "PersistedEntitiesMetadata",
+            entitiesInput,
+          )
+        : undefined;
       const entities = inputEntities
         ? await mapActionInputEntitiesToEntities({
             actorId: userAuthentication.actorId,
@@ Expand Down @@

apps/hash-ai-worker-ts/src/activities/flow-activities/infer-entities-from-content-action.ts

            
                      Original file line number
                      Diff line number
                      Diff line change
                  
    @@ -12,6 +12,10 @@ import {
  
    } from "@blockprotocol/type-system";

    import { typedKeys } from "@local/advanced-types/typed-entries";

    import type { AiFlowActionActivity } from "@local/hash-backend-utils/flows";

    import {

      getStorageProvider,

      storePayload,

    } from "@local/hash-backend-utils/flows/payload-storage";

    import { isInferenceModelName } from "@local/hash-isomorphic-utils/ai-inference-types";

    import { getSimplifiedAiFlowActionInputs } from "@local/hash-isomorphic-utils/flows/action-definitions";

    import type { ProposedEntity } from "@local/hash-isomorphic-utils/flows/types";

    @@ -43,7 +47,7 @@ export const inferEntitiesFromContentAction: AiFlowActionActivity<
  
        actionType: "inferEntitiesFromContent",

      });

      const { flowEntityId, userAuthentication, stepId, webId } =

      const { flowEntityId, userAuthentication, stepId, webId, workflowId, runId } =

        await getFlowContext();

      const aiAssistantAccountId = await getAiAssistantAccountIdActivity({

    @@ -208,6 +212,17 @@ export const inferEntitiesFromContentAction: AiFlowActionActivity<
  
        }),

      );

      // Store the proposed entities in S3 to avoid passing large payloads through Temporal

      const storedRef = await storePayload({

        storageProvider: getStorageProvider(),

        workflowId,

        runId,

        stepId,

        outputName: "proposedEntities",

        kind: "ProposedEntity",

        value: proposedEntities,

      });

      return {

        code: StatusCode.Ok,

        contents: [

    @@ -217,7 +232,7 @@ export const inferEntitiesFromContentAction: AiFlowActionActivity<
  
                outputName: "proposedEntities",

                payload: {

                  kind: "ProposedEntity",

                  value: proposedEntities,

                  value: storedRef,

                },

              },

            ],

apps/hash-ai-worker-ts/src/activities/flow-activities/infer-metadata-from-document-action.ts

-Original file line number
+Diff line change
@@ Expand Up / @@ -7,6 +7,10 @@ import type { @@
     } from "@blockprotocol/type-system";
     import { extractEntityUuidFromEntityId } from "@blockprotocol/type-system";
     import type { AiFlowActionActivity } from "@local/hash-backend-utils/flows";
+    import {
+      getStorageProvider,
+      storePayload,
+    } from "@local/hash-backend-utils/flows/payload-storage";
     import type { HashEntity } from "@local/hash-graph-sdk/entity";
     import { getSimplifiedAiFlowActionInputs } from "@local/hash-isomorphic-utils/flows/action-definitions";
     import type { PersistedEntityMetadata } from "@local/hash-isomorphic-utils/flows/types";
@@ Expand Down Expand Up @@
     > = async ({ inputs }) => {
       const {
         flowEntityId,
+        runId,
         stepId,
         userAuthentication: { actorId: userActorId },
         webId,
+        workflowId,
       } = await getFlowContext();
       const { documentEntityId } = getSimplifiedAiFlowActionInputs({
@@ Expand Down Expand Up @@
           propertyProvenance,
         });
+      // Store the proposed entities in S3 to avoid passing large payloads through Temporal
+      const storedRef = await storePayload({
+        storageProvider: getStorageProvider(),
+        workflowId,
+        runId,
+        stepId,
+        outputName: "proposedEntities",
+        kind: "ProposedEntity",
+        value: proposedEntities,
+      });
       return {
         code: StatusCode.Ok,
         contents: [
@@ Expand All @@
                 outputName: "proposedEntities",
                 payload: {
                   kind: "ProposedEntity",
-                  value: proposedEntities,
+                  value: storedRef,
                 },
               },
               {
@@ Expand Down @@

apps/hash-ai-worker-ts/src/activities/flow-activities/persist-entities-action.ts

-Original file line number
+Diff line change
@@ -1,5 +1,10 @@
     import type { EntityId } from "@blockprotocol/type-system";
     import type { AiFlowActionActivity } from "@local/hash-backend-utils/flows";
+    import {
+      getStorageProvider,
+      resolvePayloadValue,
+      storePayload,
+    } from "@local/hash-backend-utils/flows/payload-storage";
     import { flattenPropertyMetadata } from "@local/hash-graph-sdk/entity";
     import { getSimplifiedAiFlowActionInputs } from "@local/hash-isomorphic-utils/flows/action-definitions";
     import type {
@@ Expand All / @@ -8,19 +13,27 @@ import type { @@
       ProposedEntityWithResolvedLinks,
     } from "@local/hash-isomorphic-utils/flows/types";
     import { StatusCode } from "@local/status";
+    import { Context } from "@temporalio/activity";
-    import {
-      fileEntityTypeIds,
-      persistEntityAction,
-    } from "./persist-entity-action.js";
+    import { getFlowContext } from "../shared/get-flow-context.js";
+    import { fileEntityTypeIds, persistEntity } from "./persist-entity-action.js";
     export const persistEntitiesAction: AiFlowActionActivity<
       "persistEntities"
     > = async ({ inputs }) => {
-      const { draft, proposedEntities } = getSimplifiedAiFlowActionInputs({
-        inputs,
-        actionType: "persistEntities",
-      });
+      const { runId, stepId, workflowId } = await getFlowContext();
+      const { draft, proposedEntities: proposedEntitiesInput } =
+        getSimplifiedAiFlowActionInputs({
+          inputs,
+          actionType: "persistEntities",
+        });
+      const proposedEntities = await resolvePayloadValue(
+        getStorageProvider(),
+        "ProposedEntity",
+        proposedEntitiesInput,
+      );
       /**
        * Sort the entities to persist in dependency order:
@@ Expand Down Expand Up / @@ -78,6 +91,9 @@ export const persistEntitiesAction: AiFlowActionActivity< @@
        * if an existing entity is found to update rather than a new one with the localId being created.
        */
       for (const unresolvedEntity of entitiesWithDependenciesSortedLast) {
+        // Heartbeat to indicate the activity is still running
+        Context.current().heartbeat();
         const {
           claims,
           entityTypeIds,
@@ Expand Down Expand Up / @@ -158,20 +174,9 @@ export const persistEntitiesAction: AiFlowActionActivity< @@
           }
         }
-        const persistedEntityOutputs = await persistEntityAction({
-          inputs: [
-            {
-              inputName: "draft",
-              payload: { kind: "Boolean", value: draft ?? false },
-            },
-            {
-              inputName: "proposedEntityWithResolvedLinks",
-              payload: {
-                kind: "ProposedEntityWithResolvedLinks",
-                value: entityWithResolvedLinks,
-              },
-            },
-          ],
+        const persistedEntityOutputs = await persistEntity({
+          proposedEntityWithResolvedLinks: entityWithResolvedLinks,
+          draft: draft ?? false,
         });
         const output = persistedEntityOutputs.contents[0]?.outputs[0]?.payload;
@@ Expand Down Expand Up / @@ -210,6 +215,20 @@ export const persistEntitiesAction: AiFlowActionActivity< @@
       const persistedEntities = Object.values(persistedEntitiesByLocalId);
+      // Store the output in S3 to avoid passing large payloads through Temporal
+      const storedRef = await storePayload({
+        storageProvider: getStorageProvider(),
+        workflowId,
+        runId,
+        stepId,
+        outputName: "persistedEntities",
+        kind: "PersistedEntitiesMetadata",
+        value: {
+          persistedEntities,
+          failedEntityProposals: Object.values(failedEntitiesByLocalId),
+        },
+      });
       return {
         /** @todo H-2604 have some kind of 'partially completed' status when reworking flow return codes */
         code:
@@ Expand All / @@ -231,10 +250,7 @@ export const persistEntitiesAction: AiFlowActionActivity< @@
                 outputName: "persistedEntities",
                 payload: {
                   kind: "PersistedEntitiesMetadata",
-                  value: {
-                    persistedEntities,
-                    failedEntityProposals: Object.values(failedEntitiesByLocalId),
-                  },
+                  value: storedRef,
                 },
               },
             ],
@@ Expand Down @@

apps/hash-ai-worker-ts/src/activities/flow-activities/persist-entity-action.ts

-Original file line number
+Diff line change
@@ -1,6 +1,10 @@
     import type { EntityId, VersionedUrl } from "@blockprotocol/type-system";
     import { extractEntityUuidFromEntityId } from "@blockprotocol/type-system";
     import type { AiFlowActionActivity } from "@local/hash-backend-utils/flows";
+    import {
+      getStorageProvider,
+      resolvePayloadValue,
+    } from "@local/hash-backend-utils/flows/payload-storage";
     import { getWebMachineId } from "@local/hash-backend-utils/machine-actors";
     import type { CreateEntityParameters } from "@local/hash-graph-sdk/entity";
     import {
@@ Expand All / @@ -9,7 +13,10 @@ import { @@
       mergePropertyObjectAndMetadata,
     } from "@local/hash-graph-sdk/entity";
     import { getSimplifiedAiFlowActionInputs } from "@local/hash-isomorphic-utils/flows/action-definitions";
-    import type { PersistedEntityMetadata } from "@local/hash-isomorphic-utils/flows/types";
+    import type {
+      PersistedEntityMetadata,
+      ProposedEntityWithResolvedLinks,
+    } from "@local/hash-isomorphic-utils/flows/types";
     import { systemEntityTypes } from "@local/hash-isomorphic-utils/ontology-type-ids";
     import type {
       HasObject,
@@ Expand Down Expand Up / @@ -47,23 +54,26 @@ export const fileEntityTypeIds: VersionedUrl[] = [ @@
       systemEntityTypes.pptxPresentation.entityTypeId,
     ];
-    export const persistEntityAction: AiFlowActionActivity<
-      "persistEntity"
-    > = async ({ inputs }) => {
+    /**
+     * Inner function that handles the actual entity persistence logic.
+     * This is called by both persistEntityAction (which resolves the payload ref first)
+     * and persistEntitiesAction (which passes the resolved value directly).
+     */
+    export const persistEntity = async ({
+      proposedEntityWithResolvedLinks,
+      draft,
+    }: {
+      proposedEntityWithResolvedLinks: ProposedEntityWithResolvedLinks;
+      draft: boolean;
+    }): Promise<ReturnType<AiFlowActionActivity<"persistEntity">>> => {
       const {
         flowEntityId,
         stepId,
         userAuthentication: { actorId },
         webId,
       } = await getFlowContext();
-      const { draft, proposedEntityWithResolvedLinks } =
-        getSimplifiedAiFlowActionInputs({
-          inputs,
-          actionType: "persistEntity",
-        });
-      const createEditionAsDraft = draft ?? false;
+      const createEditionAsDraft = draft;
       const {
         entityTypeIds,
@@ Expand Down Expand Up / @@ -336,3 +346,27 @@ export const persistEntityAction: AiFlowActionActivity< @@
         ],
       };
     };
+    /**
+     * Flow action activity that persists a single entity.
+     */
+    export const persistEntityAction: AiFlowActionActivity<
+      "persistEntity"
+    > = async ({ inputs }) => {
+      const { draft, proposedEntityWithResolvedLinks: proposedEntityInput } =
+        getSimplifiedAiFlowActionInputs({
+          inputs,
+          actionType: "persistEntity",
+        });
+      const proposedEntityWithResolvedLinks = await resolvePayloadValue(
+        getStorageProvider(),
+        "ProposedEntityWithResolvedLinks",
+        proposedEntityInput,
+      );
+      return persistEntity({
+        proposedEntityWithResolvedLinks,
+        draft: draft ?? false,
+      });
+    };

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

H-3848, H-6154: Handle larger Flow payloads, historical flight data Flows #8356

Uh oh!

Diff view

Diff view

There are no files selected for viewing

Uh oh!

Uh oh!

Uh oh!

Uh oh!