USE 597 - Support new parallel enrichments block in StepFunction

[ghukill]

Purpose and background context

Warning

These changes must be made in lockstep with StepFunction changes outlined in PR for USE-560

Why these changes are being introduced:

The requirement going into this work was the ability to orchestrate the harvesting of fulltext for DSpace theses. A new harvester has been built and deployed already, and the general pattern could have been quite similar to embeddings, just daisy-chained after embeddings work in the StepFunction.

However, this provided an opportunity to make some updates and improvements in the StepFunction. Previously, we had no way to skip embeddings for a given ETL run. With the addition of fulltext harvesting, which naively would have landed after the embeddings block in the StepFunction, this ability to skip particular stages would have been increasingly difficult to manage.

Additionally, from a performance POV, it was not ideal to have the embeddings and fulltext harvesting work running sequentially when there is no reason they could not run in parallel.

In summary, the need to support fulltext harvesting as new functionality in the StepFunction provided a good time to make some structural updates.

How this addresses that need:

The changes in the StepFunction are pretty significant, but often that requires only small alignments in the pipeline lambda. These changes are partially alignment with the StepFunction and truly new handlers for the dspace fulltext harvesting + indexing.

• New handlers are added for new states in the TIMDEX StepFunction

  • prepare an "enrichments" object that provides a way to skip specific enrichments
  • prepare CLI commands for DSpace fulltext harvester and TIM indexing of fulltexts
  • manage the results of the parallel state in the StepFunction, resulting in a final success true/false

• Continued refinement of next-step = (exit-ok, exit-error, end)

  • branches in the Parallel block in the StepFunction now return "end" indicating the branch is done, but this does not mean the StepFunction itself is done

• Lastly, scaffold a final ETL run metrics handler

  • currently just logging the results, but this is a great place to publish to AWS metrics if desired

How can a reviewer manually see the effects of these changes?

1- Set Dev1 TimdexMangers AWS credentials

2- Build SAM image for pipeline lambda:

make sam-build

And confirm that an env vars file is present at tests/sam/env.json:

{
  "TimdexPipelineLambda": {
  	"TIMDEX_ALMA_EXPORT_BUCKET_ID":"not-needed",
  	"TIMDEX_S3_EXTRACT_BUCKET_ID":"timdex-extract-dev-222053980223"
  }
}

Note

the following assume that jq is installed for JSON parsing in the terminal

3- Confirm that new enrichments section is propagated regardless of next-step. Here we make a call with next-step = transform, but the enrichments block persists:

echo '{
  "next-step": "transform",
  "run-date": "2026-08-18",
  "run-type": "daily",
  "source": "dspace",
  "verbose": true,
  "enrichments": {
    "message": "hello world!",
    "skip": false
  }
}' | sam local invoke --env-vars tests/sam/env.json -e - | jq

{
  "next-step": "load",
  "run-date": "2026-08-18",
  "run-type": "daily",
  "run-id": "6156c4dc-8559-4ad9-8c43-e5fba525cd01",
  "source": "dspace",
  "verbose": true,
  "transform": {
    "files-to-transform": [
      {
        "transform-command": [
          "--input-file=s3://timdex-extract-dev-222053980223/dspace/dspace-2026-08-18-daily-extracted-records-to-index.xml",
          "--output-location=s3://timdex-extract-dev-222053980223/dataset",
          "--source=dspace",
          "--run-id=6156c4dc-8559-4ad9-8c43-e5fba525cd01",
          "--run-timestamp=2026-08-25T13:53:09.567016+00:00"
        ]
      }
    ]
  },
  "enrichments": {  //<--- Note presence is maintained
    "message": "hello world!",
    "skip": false
  }
}

4- Confirm that a payload without an "enrichments" block is created by lambda handler that begins enrichments parallel work, and that next-step = finalize:

echo '{
  "next-step": "enrichment",
  "run-date": "2026-08-18",
  "run-type": "daily",
  "source": "dspace",
  "verbose": true 
}' | sam local invoke --env-vars tests/sam/env.json -e - | jq

{
  "next-step": "finalize", //<--- Next step is "finalize"
  "run-date": "2026-08-18",
  "run-type": "daily",
  "run-id": "618c6550-55d2-4c40-affd-07e4812df2d6",
  "source": "dspace",
  "verbose": true,
  "enrichments": {  //<--- Added by lambda
    "skip": false,
    "embeddings": {
      "skip": false
    },
    "fulltexts": {
      "skip": false
    }
  }
}

And, confirm that skip overrides are preserved:

echo '{
  "next-step": "enrichment",
  "run-date": "2026-08-18",
  "run-type": "daily",
  "source": "dspace",
  "verbose": true,
  "enrichments":{
    "skip": true
  }
}' | sam local invoke --env-vars tests/sam/env.json -e - | jq

{
  "next-step": "finalize",
  "run-date": "2026-08-18",
  "run-type": "daily",
  "run-id": "33d99315-f1d3-41ee-9741-ef6ff16cb4b5",
  "source": "dspace",
  "verbose": true,
  "enrichments": {
    "skip": true,  //<--- override persisted
    "embeddings": {
      "skip": false
    },
    "fulltexts": {
      "skip": false
    }
  }
}

5- Confirm that "finalize" handler in pipeline lambad retrieves run metrics. Here, we'll use a run from digitalcollections to demonstrate any run should work, no special treatment is needed.

echo '{
  "next-step": "finalize",
  "run-date": "2026-08-13",
  "run-type": "full",
  "run-id": "496a7d87-9027-4351-9673-116a1492eef7",
  "source": "digitalcollections",
  "verbose": true
}' | sam local invoke --env-vars tests/sam/env.json -e - | jq

Logging output:

2026-08-25 14:03:14,262 INFO lambdas.format_input.handle_finalize() line 507: Run metrics: {"records": {"count": 81762, "actions": {"index": 81762}}, "embeddings": {"count": 81762}, "fulltexts": {"count": 0}}

{
  "next-step": "end",
  "run-date": "2026-08-13",
  "run-type": "full",
  "run-id": "496a7d87-9027-4351-9673-116a1492eef7",
  "source": "digitalcollections",
  "verbose": true,
  "metrics": {  //<--- metrics are included in lambda response as well
    "records": {
      "count": 81762,
      "actions": {
        "index": 81762
      }
    },
    "embeddings": {
      "count": 81762
    },
    "fulltexts": {
      "count": 0
    }
  }
}

Includes new or updated dependencies?

YES

Changes expectations for external applications?

YES

• Capable of handling new StepFunction architecture outlined in USE-560
• Logging of ETL run metrics

What are the relevant tickets?

• https://mitlibraries.atlassian.net/browse/USE-597
• https://mitlibraries.atlassian.net/browse/USE-560

Code review

• Code review best practices are documented here and you are encouraged to have a constructive dialogue with your reviewers about their preferences and expectations.

[Copilot]

Pull request overview

This PR updates the pipeline input-formatting Lambda to align with a new StepFunction architecture that runs enrichments (embeddings + DSpace fulltext harvesting/loading) in parallel, while also adding a scaffolded “finalize” step that reports ETL run metrics.

Changes:

• Introduces an enrichment entry step that prepares/normalizes an enrichments payload for a StepFunctions Parallel block and routes per-branch enrichment steps.
• Adds fulltext harvesting/loading command preparation and a finalize handler that attaches run metrics to the StepFunction output.
• Adds dependencies (duckdb, pandas) and expands test coverage for new enrichment/fulltext/metrics behavior.

Reviewed changes

Copilot reviewed 8 out of 9 changed files in this pull request and generated 4 comments.

Show a summary per file

File	Description
lambdas/format_input.py	Adds enrichment routing, fulltexts handlers, and a finalize step that logs/attaches metrics.
lambdas/helpers.py	Adds get_run_metrics() for record/action/enrichment counts from the dataset metadata.
lambdas/commands.py	Adds command generators for fulltext harvest and fulltext load operations.
lambdas/config.py	Expands valid input steps for the updated StepFunction flow; adds valid fulltexts sources.
pyproject.toml	Adds duckdb and pandas dependencies and mypy overrides for missing stubs.
uv.lock	Updates the resolved dependency set to include new requirements.
tests/test_format_input.py	Updates expectations for new enrichment step and adds tests for fulltexts + finalize output.
tests/test_helpers.py	Adds unit tests for get_run_metrics() including missing-view behavior.
tests/test_commands.py	Adds unit tests for fulltext command generation helpers.

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[ehanson8]

Looks good to me and works as expected, one question and an optional suggestion but otherwise approved!