From e14cd1ca7677d6cc0fb8d67094c6edc1c860f99e Mon Sep 17 00:00:00 2001 From: Ling-Sen Peng Date: Wed, 12 Aug 2026 14:29:16 -0700 Subject: [PATCH 1/3] Stop asserting the server's exact 404 wording for a deleted workflow test_all asserted the deleted-workflow 404 read exactly "workflow with id: not found." The server now returns "No execution found for id: ", so the test fails on every branch (reproduced on fix/pin-mcp-below-2 and fix/13-hierarchical-agents-demo; main last went green Aug 7, before the change). Assert the durable part instead: a 404 whose message names the execution. --- tests/integration/client/orkes/test_orkes_clients.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/tests/integration/client/orkes/test_orkes_clients.py b/tests/integration/client/orkes/test_orkes_clients.py index 2811d492..173a1cce 100644 --- a/tests/integration/client/orkes/test_orkes_clients.py +++ b/tests/integration/client/orkes/test_orkes_clients.py @@ -579,7 +579,10 @@ def __test_workflow_execution_lifecycle(self): workflow = self.workflow_client.get_workflow(workflow_uuid, False) except ApiException as e: assert e.code == 404 - assert str(e.message).lower() == "workflow with id: {} not found.".format(workflow_uuid) + # The server's wording for this 404 changes across releases + # ("Workflow with Id: X not found." -> "No execution found for id: X"), + # so assert on the durable part: it is a 404 naming this execution. + assert workflow_uuid in str(e.message) def __test_task_execution_lifecycle(self): From 9c9bdbb777e022a3b5381a1c05a0bb80cdf90779 Mon Sep 17 00:00:00 2001 From: Ling-Sen Peng Date: Wed, 12 Aug 2026 14:38:39 -0700 Subject: [PATCH 2/3] Unregister the per-run task defs the integration suites create MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The sdkdev account has hit its cap: 402 System has reached the maximum allowed Task Definitions limit of 1000. so registration now fails for every branch, and test_05_verify_task_definitions 404s on the def that could not be registered. Three suites register RUN_ID-suffixed task defs via worker_task(register_task_def=True) and never remove them, leaking ~11 per run across all four integration jobs: test_comprehensive_e2e (5 tasks + 1 workflow), test_async_lease_extension (4 + 4), test_lease_extension (2 + 2, which had no tearDownClass at all). Cleanup is best-effort — a failed unregister warns rather than reddening a passing suite. scripts/prune_leaked_test_task_defs.py clears the backlog already on the server; this stops the leak but cannot free the 1000 defs already there. It is a dry run unless given --delete, and only matches the suites' own prefixes followed by a run id, so no hand-registered def is a candidate. --- scripts/prune_leaked_test_task_defs.py | 105 ++++++++++++++++++ tests/integration/conftest.py | 38 +++++++ .../integration/test_async_lease_extension.py | 11 ++ tests/integration/test_comprehensive_e2e.py | 6 + tests/integration/test_lease_extension.py | 12 ++ 5 files changed, 172 insertions(+) create mode 100644 scripts/prune_leaked_test_task_defs.py diff --git a/scripts/prune_leaked_test_task_defs.py b/scripts/prune_leaked_test_task_defs.py new file mode 100644 index 00000000..1c4785e5 --- /dev/null +++ b/scripts/prune_leaked_test_task_defs.py @@ -0,0 +1,105 @@ +#!/usr/bin/env python3 +"""Delete task definitions left behind by integration-test runs. + +The integration suites register per-run task defs (``sync_basic_``, +``async_lease_heartbeat_task_``, ...) and, before the tearDownClass +cleanup was added, never removed them. On a shared server those accumulate +until the account hits its Task Definitions cap, at which point every +registration answers:: + + 402 System has reached the maximum allowed Task Definitions limit of 1000. + +and the integration jobs fail on unrelated branches. This prunes the +leftovers so the cap has room again. + +Dry run by default — it prints what it would delete and exits. Pass --delete +to actually remove them. Reads the usual CONDUCTOR_SERVER_URL / +CONDUCTOR_AUTH_KEY / CONDUCTOR_AUTH_SECRET environment. + + python scripts/prune_leaked_test_task_defs.py # list matches + python scripts/prune_leaked_test_task_defs.py --delete # remove them +""" + +import argparse +import re +import sys + +from conductor.client.configuration.configuration import Configuration +from conductor.client.orkes.orkes_metadata_client import OrkesMetadataClient + +# Prefixes owned by the integration suites, each followed by a per-run id. +# Only names matching one of these AND ending in a run id are touched, so a +# hand-registered or production task def is never a candidate. +TEST_PREFIXES = ( + # tests/integration/test_comprehensive_e2e.py + "sync_basic_", + "async_basic_", + "complex_schema_", + "task_in_progress_", + "failing_task_", + # tests/integration/test_lease_extension.py + "lease_heartbeat_task_", + "lease_no_heartbeat_task_", + # tests/integration/test_async_lease_extension.py + "async_lease_heartbeat_task_", + "async_lease_no_heartbeat_task_", + "async_lease_fast_with_hb_", + "async_lease_fast_no_hb_", + # tests/integration/client/orkes/test_orkes_clients.py (shortuuid suffix) + "IntegrationTestOrkesClientsTask_", +) + +# uuid4().hex[:8] for most suites; shortuuid for test_orkes_clients. +RUN_ID = re.compile(r"^(?:[0-9a-f]{8}|[0-9A-Za-z]{20,25})$") + + +def is_leaked(name): + for prefix in TEST_PREFIXES: + if name.startswith(prefix) and RUN_ID.match(name[len(prefix):]): + return True + return False + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--delete", + action="store_true", + help="actually unregister the matches (default: dry run)", + ) + args = parser.parse_args() + + config = Configuration() + client = OrkesMetadataClient(config) + + all_defs = client.get_all_task_defs() + leaked = sorted(d.name for d in all_defs if is_leaked(d.name)) + + print(f"server: {config.host}") + print(f"task defs total: {len(all_defs)}") + print(f"test leftovers: {len(leaked)}") + + if not leaked: + return 0 + + if not args.delete: + for name in leaked: + print(f" would delete {name}") + print("\nDry run — re-run with --delete to remove these.") + return 0 + + failed = 0 + for name in leaked: + try: + client.unregister_task_def(name) + print(f" deleted {name}") + except Exception as e: + failed += 1 + print(f" FAILED {name}: {e}", file=sys.stderr) + + print(f"\ndeleted {len(leaked) - failed} of {len(leaked)}") + return 1 if failed else 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index aede140e..9b7e9b3e 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -70,6 +70,44 @@ def setUpClass(cls): raise unittest.SkipTest(reason) +# --------------------------------------------------------------------------- +# Metadata cleanup +# --------------------------------------------------------------------------- + +def cleanup_metadata(config, task_defs=(), workflow_defs=()): + """Best-effort delete of metadata a test class registered. + + Suites that register per-run task defs (``worker_task(..., + register_task_def=True)`` with a RUN_ID-suffixed name) used to leave every + one behind, so each CI run added a handful to the shared server until it + hit its Task Definitions cap and answered 402 to all further + registrations. Call this from tearDownClass. + + Failures are logged and swallowed: cleanup must never turn a passing suite + red, and a def that was never registered is nothing to report. + + ``workflow_defs`` items are names (version 1 assumed) or (name, version). + """ + from conductor.client.orkes.orkes_metadata_client import OrkesMetadataClient + + client = OrkesMetadataClient(config) + + for name in task_defs: + try: + client.unregister_task_def(name) + except Exception as e: + logger.warning("cleanup: could not unregister task def %s: %s", name, e) + + for entry in workflow_defs: + name, version = entry if isinstance(entry, tuple) else (entry, 1) + try: + client.unregister_workflow_def(name, version) + except Exception as e: + logger.warning( + "cleanup: could not unregister workflow def %s v%s: %s", name, version, e + ) + + # --------------------------------------------------------------------------- # Pytest session-scoped fixtures # --------------------------------------------------------------------------- diff --git a/tests/integration/test_async_lease_extension.py b/tests/integration/test_async_lease_extension.py index d384ac4d..b6829fbe 100644 --- a/tests/integration/test_async_lease_extension.py +++ b/tests/integration/test_async_lease_extension.py @@ -232,6 +232,17 @@ def tearDownClass(cls): handler = getattr(cls, '_task_handler', None) if handler is not None: handler.stop_processes() + from tests.integration.conftest import cleanup_metadata + cleanup_metadata( + cls.config, + task_defs=(HEARTBEAT_TASK, NO_HEARTBEAT_TASK, FAST_HB_TASK, FAST_NO_HB_TASK), + workflow_defs=( + f'test_async_lease_heartbeat_{RUN_ID}', + f'test_async_lease_no_heartbeat_{RUN_ID}', + f'test_async_perf_with_hb_{RUN_ID}', + f'test_async_perf_no_hb_{RUN_ID}', + ), + ) def _register_workflow(self, wf_name, task_names): """Register a workflow with one or more tasks in sequence.""" diff --git a/tests/integration/test_comprehensive_e2e.py b/tests/integration/test_comprehensive_e2e.py index 4fcfac8e..178c09f6 100644 --- a/tests/integration/test_comprehensive_e2e.py +++ b/tests/integration/test_comprehensive_e2e.py @@ -586,6 +586,12 @@ def tearDownClass(cls): if os.path.exists(cls.metrics_dir): import shutil shutil.rmtree(cls.metrics_dir) + from tests.integration.conftest import cleanup_metadata + cleanup_metadata( + cls.config, + task_defs=cls.EXPECTED_WORKERS, + workflow_defs=(WF_NAME,), + ) print("\n✓ Cleanup complete") diff --git a/tests/integration/test_lease_extension.py b/tests/integration/test_lease_extension.py index 3b986409..82f1aca4 100644 --- a/tests/integration/test_lease_extension.py +++ b/tests/integration/test_lease_extension.py @@ -146,6 +146,18 @@ def setUpClass(cls): cls.metadata_client = OrkesMetadataClient(cls.config) cls.workflow_client = OrkesWorkflowClient(cls.config) + @classmethod + def tearDownClass(cls): + from tests.integration.conftest import cleanup_metadata + cleanup_metadata( + cls.config, + task_defs=(HEARTBEAT_TASK, NO_HEARTBEAT_TASK), + workflow_defs=( + f'test_lease_heartbeat_{RUN_ID}', + f'test_lease_no_heartbeat_{RUN_ID}', + ), + ) + def _register_workflow(self, wf_name, task_name): """Register a single-task workflow.""" workflow = WorkflowDef(name=wf_name, version=1) From 071131e869f6094481e36652e51e1424a601835d Mon Sep 17 00:00:00 2001 From: Ling-Sen Peng Date: Wed, 12 Aug 2026 15:00:09 -0700 Subject: [PATCH 3/3] Reclaim leaked task-def quota at session start MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Unregistering on teardown stops the leak but cannot free the ~1000 defs already registered, so the account stays at its cap and every branch keeps getting 402 on registration — no run gets far enough to clean up after itself. Prune the stale leftovers before any test registers. Only names matching an integration suite's prefix followed by a run id are candidates, and only ones older than 2h: four buckets run in parallel against one server, so a def a concurrent run may still be using is off limits (as is one with no createTime, whose age is unknown). The reclaim prints how many it freed, so a run that is still capped says so instead of failing obscurely. Verified against a local server: reclaim deletes a registered sync_basic_ and reports the count; test_comprehensive_e2e (8 passed), test_lease_extension and test_async_lease_extension pass with cleanup leaving nothing behind; the deleted-workflow 404 assert holds on this server's third wording of that message ("No such workflow found by id: "). scripts/prune_leaked_test_task_defs.py now shares the matcher rather than duplicating it, and grew --include-recent for pruning by hand when no run is in flight. --- scripts/prune_leaked_test_task_defs.py | 61 ++++++-------- tests/integration/conftest.py | 55 +++++++++++- tests/integration/leaked_task_defs.py | 112 +++++++++++++++++++++++++ 3 files changed, 187 insertions(+), 41 deletions(-) create mode 100644 tests/integration/leaked_task_defs.py diff --git a/scripts/prune_leaked_test_task_defs.py b/scripts/prune_leaked_test_task_defs.py index 1c4785e5..58d0f0f2 100644 --- a/scripts/prune_leaked_test_task_defs.py +++ b/scripts/prune_leaked_test_task_defs.py @@ -9,11 +9,14 @@ 402 System has reached the maximum allowed Task Definitions limit of 1000. -and the integration jobs fail on unrelated branches. This prunes the -leftovers so the cap has room again. +and the integration jobs fail on unrelated branches. -Dry run by default — it prints what it would delete and exits. Pass --delete -to actually remove them. Reads the usual CONDUCTOR_SERVER_URL / +The suites now prune stale leftovers themselves at session start (see +tests/integration/leaked_task_defs.py), so this script is for pruning by hand +— including the defs too recent for the automatic pass to touch. + +Dry run by default: it prints what it would delete and exits. Pass --delete to +actually remove them. Reads the usual CONDUCTOR_SERVER_URL / CONDUCTOR_AUTH_KEY / CONDUCTOR_AUTH_SECRET environment. python scripts/prune_leaked_test_task_defs.py # list matches @@ -21,44 +24,16 @@ """ import argparse -import re import sys from conductor.client.configuration.configuration import Configuration from conductor.client.orkes.orkes_metadata_client import OrkesMetadataClient - -# Prefixes owned by the integration suites, each followed by a per-run id. -# Only names matching one of these AND ending in a run id are touched, so a -# hand-registered or production task def is never a candidate. -TEST_PREFIXES = ( - # tests/integration/test_comprehensive_e2e.py - "sync_basic_", - "async_basic_", - "complex_schema_", - "task_in_progress_", - "failing_task_", - # tests/integration/test_lease_extension.py - "lease_heartbeat_task_", - "lease_no_heartbeat_task_", - # tests/integration/test_async_lease_extension.py - "async_lease_heartbeat_task_", - "async_lease_no_heartbeat_task_", - "async_lease_fast_with_hb_", - "async_lease_fast_no_hb_", - # tests/integration/client/orkes/test_orkes_clients.py (shortuuid suffix) - "IntegrationTestOrkesClientsTask_", +from tests.integration.leaked_task_defs import ( + STALE_AFTER_SECONDS, + is_leaked_task_def, + stale_leaked_task_defs, ) -# uuid4().hex[:8] for most suites; shortuuid for test_orkes_clients. -RUN_ID = re.compile(r"^(?:[0-9a-f]{8}|[0-9A-Za-z]{20,25})$") - - -def is_leaked(name): - for prefix in TEST_PREFIXES: - if name.startswith(prefix) and RUN_ID.match(name[len(prefix):]): - return True - return False - def main(): parser = argparse.ArgumentParser(description=__doc__) @@ -67,13 +42,25 @@ def main(): action="store_true", help="actually unregister the matches (default: dry run)", ) + parser.add_argument( + "--include-recent", + action="store_true", + help=( + "also prune defs newer than " + f"{STALE_AFTER_SECONDS // 3600}h — only safe when no integration " + "run is in flight, since a concurrent run's defs are fair game" + ), + ) args = parser.parse_args() config = Configuration() client = OrkesMetadataClient(config) all_defs = client.get_all_task_defs() - leaked = sorted(d.name for d in all_defs if is_leaked(d.name)) + if args.include_recent: + leaked = sorted(d.name for d in all_defs if is_leaked_task_def(d.name)) + else: + leaked = stale_leaked_task_defs(all_defs) print(f"server: {config.host}") print(f"task defs total: {len(all_defs)}") diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index 9b7e9b3e..3f7de2a7 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -52,6 +52,36 @@ def _check_server_connectivity(): return _server_available, _skip_reason +_reclaimed = False + + +def _reclaim_once(): + """Reclaim task-def quota leaked by earlier runs, before any test registers. + + Without this a filled-up account stays filled: registration answers 402 for + every branch, so no run gets far enough to clean up after itself. + + Called from both entry points below — pytest_sessionstart covers the CI + invocations (tests/integration/conftest.py is an initial conftest for every + bucket), skip_if_server_unavailable covers a session that reaches the + unittest suites some other way. The flag makes the second call a no-op. + """ + global _reclaimed + if _reclaimed: + return + _reclaimed = True + + available, _ = _check_server_connectivity() + if not available: + return + from tests.integration.leaked_task_defs import reclaim_task_def_quota + reclaim_task_def_quota(Configuration()) + + +def pytest_sessionstart(session): + _reclaim_once() + + def skip_if_server_unavailable(): """ Call from unittest.TestCase.setUpClass to skip the entire test class @@ -92,20 +122,37 @@ def cleanup_metadata(config, task_defs=(), workflow_defs=()): client = OrkesMetadataClient(config) + def _log(what, e): + # "no such definition" is the normal case for anything a deselected or + # skipped test never registered, so it logs at debug — only a cleanup + # that failed for some other reason is worth a warning. + level = logging.DEBUG if _is_missing(e) else logging.WARNING + logger.log(level, "cleanup: could not unregister %s: %s", what, e) + for name in task_defs: try: client.unregister_task_def(name) except Exception as e: - logger.warning("cleanup: could not unregister task def %s: %s", name, e) + _log(f"task def {name}", e) for entry in workflow_defs: name, version = entry if isinstance(entry, tuple) else (entry, 1) try: client.unregister_workflow_def(name, version) except Exception as e: - logger.warning( - "cleanup: could not unregister workflow def %s v%s: %s", name, version, e - ) + _log(f"workflow def {name} v{version}", e) + + +def _is_missing(exc): + """True when a delete failed only because the definition was not there. + + The server reports this inconsistently — 404, or a 500 whose body reads + "No such task definition" / "No such workflow definition" — so match on + both the status and the text. + """ + if getattr(exc, "code", None) == 404: + return True + return "no such" in str(getattr(exc, "body", "") or exc).lower() # --------------------------------------------------------------------------- diff --git a/tests/integration/leaked_task_defs.py b/tests/integration/leaked_task_defs.py new file mode 100644 index 00000000..098ca415 --- /dev/null +++ b/tests/integration/leaked_task_defs.py @@ -0,0 +1,112 @@ +"""Recognise and reclaim task definitions left behind by integration runs. + +The integration suites register per-run task defs +(``sync_basic_``, ``async_lease_heartbeat_task_``, ...). Until +the ``tearDownClass`` cleanup landed, every run left its own behind, and on the +shared server they accumulated until the account hit its cap:: + + 402 System has reached the maximum allowed Task Definitions limit of 1000. + +At that point registration fails for every branch, so unrelated PRs go red. +Cleanup on teardown stops the leak; this module reclaims what earlier runs +already leaked, and is also what ``scripts/prune_leaked_test_task_defs.py`` +matches on. +""" + +import logging +import os +import re +import time + +logger = logging.getLogger(__name__) + +# Prefixes owned by the integration suites, each followed by a per-run id. +# A name is only a candidate if it matches one of these AND ends in a run id, +# so a hand-registered or production task def is never touched. +TEST_PREFIXES = ( + # tests/integration/test_comprehensive_e2e.py + "sync_basic_", + "async_basic_", + "complex_schema_", + "task_in_progress_", + "failing_task_", + # tests/integration/test_lease_extension.py + "lease_heartbeat_task_", + "lease_no_heartbeat_task_", + # tests/integration/test_async_lease_extension.py + "async_lease_heartbeat_task_", + "async_lease_no_heartbeat_task_", + "async_lease_fast_with_hb_", + "async_lease_fast_no_hb_", + # tests/integration/client/orkes/test_orkes_clients.py (shortuuid suffix) + "IntegrationTestOrkesClientsTask_", +) + +# uuid4().hex[:8] for most suites; shortuuid for test_orkes_clients. +_RUN_ID = re.compile(r"^(?:[0-9a-f]{8}|[0-9A-Za-z]{20,25})$") + +# Four integration jobs run in parallel, and other PRs run at the same time. +# Only defs older than this are reclaimed, so a concurrent run never has the +# task def it is mid-test on deleted from under it. +STALE_AFTER_SECONDS = 2 * 60 * 60 + + +def is_leaked_task_def(name): + """True if ``name`` is a per-run task def owned by the integration suites.""" + for prefix in TEST_PREFIXES: + if name.startswith(prefix) and _RUN_ID.match(name[len(prefix):]): + return True + return False + + +def stale_leaked_task_defs(task_defs, now=None): + """Names in ``task_defs`` that are leaked AND older than STALE_AFTER_SECONDS. + + A def with no ``create_time`` is left alone: without an age there is no way + to tell it from one a concurrent run just registered. + """ + cutoff_ms = ((now if now is not None else time.time()) - STALE_AFTER_SECONDS) * 1000 + return sorted( + d.name + for d in task_defs + if is_leaked_task_def(d.name) and (d.create_time or 0) and d.create_time < cutoff_ms + ) + + +def reclaim_task_def_quota(config): + """Delete stale leaked task defs so registration has room again. + + Best-effort and never raises: this runs before the suites do, and a server + that will not answer is the tests' problem to report, not this helper's. + Returns the number deleted. Set CONDUCTOR_SKIP_TASK_DEF_RECLAIM=1 to skip. + """ + if os.environ.get("CONDUCTOR_SKIP_TASK_DEF_RECLAIM"): + return 0 + + from conductor.client.orkes.orkes_metadata_client import OrkesMetadataClient + + client = OrkesMetadataClient(config) + try: + all_defs = client.get_all_task_defs() + except Exception as e: + logger.warning("reclaim: could not list task defs: %s", e) + return 0 + + stale = stale_leaked_task_defs(all_defs) + if not stale: + return 0 + + deleted = 0 + for name in stale: + try: + client.unregister_task_def(name) + deleted += 1 + except Exception as e: + logger.warning("reclaim: could not unregister %s: %s", name, e) + + # print(), not logger: logging is not configured yet at session start. + print( + f"reclaimed {deleted} of {len(stale)} stale test task defs " + f"({len(all_defs)} defs on {config.host} before pruning)" + ) + return deleted