test: cover instance queries and manual index preparation

api-tests/attribute-index-preparation/README.md

@@ -0,0 +1,69 @@
+# Manual attribute index preparation (current contract)
+
+Publication creates no index jobs or generated columns. `wf indexes generate` writes an offline SQL batch;
+the DBA executes it in a maintenance window. The former automatic scenario below is historical.
+
+Run `test_manual_indexes.py` against the locally built worktree runtime and an isolated test domain:
+
+```sh
+VNEXT_BASE_URL=http://localhost:<port> \
+VNEXT_TEST_DOMAIN=<isolated-domain> \
+VNEXT_TEST_PG_CONTAINER=<isolated-postgres> \
+VNEXT_TEST_DATABASE=<isolated-database> \
+VNEXT_ALLOW_ISOLATED_TEST_DDL=true \
+python3 api-tests/attribute-index-preparation/test_manual_indexes.py
+```
+
+`NODE_BINARY` and `VNEXT_CLI_ROOT` may override the local Node executable and sibling CLI checkout.
+The harness uses a unique flow, executes generated SQL only on that fixture, and leaves it for inspection.
+It never modifies pre-existing workflow schemas. The CLI itself never contacts the database/API.
+
+Verified 2026-09-14: **4 checks passed**, local Orchestration built from the dirty worktree, PostgreSQL 18.6,
+`autoindexdirect` isolated domain on localhost:4301. Covered: no automatic work on first publication;
+CLI generation leaves DB unchanged; explicit SQL creates ready catalog and replay preserves OIDs;
+subsequent indexed Master update creates no jobs and leaves catalog unchanged.
+
+## Historical automatic experiment (superseded)
+
+# Automatic attribute index preparation
+
+Run against an isolated **locally built** runtime with automatic preparation enabled. The selected `directly: true` path works
+with the existing background-job poller disabled. The script publishes uniquely named components through HTTP.
+PostgreSQL is inspected with read-only SQL; one read lock is held to prove the no-op path does
+not request ACCESS EXCLUSIVE. No direct optimization DDL is performed by the test.
+
+```sh
+VNEXT_BASE_URL=http://localhost:4301 \
+VNEXT_TEST_DOMAIN=autoindexdirect \
+VNEXT_TEST_PG_CONTAINER=auto-index-pg \
+VNEXT_TEST_DATABASE=vNext_autoindex_direct \
+python3 api-tests/attribute-index-preparation/test_auto_indexes.py
+```
+
+Expected checks: unindexed schema; Master package update affects latest but not pinned flow;
+trigram/B-tree generation; no-op while a conflicting read lock is held; conflict/exception rollback;
+workflow reference update. Job status and retry assertions require a dedicated clean job database.
+
+## 2026-09-11 result: blocked, not passed
+
+Worktree runtime on `localhost:4301`, PostgreSQL 18.6, Dapr 1.16.19, isolated Redis/Scheduler.
+Some initial jobs completed and created projections. Immediate Dapr callbacks also arrived
+before Aether's arming confirmation committed; dispatcher acknowledged them as `not found in
+an active state`, then their rows became Scheduled and stayed there. The scenario timed out
+at its first job-completion gate. Later checks are **not verified**.
+
+Aether 1.0.39 SDK changes require separate authorization under the runtime repository policy.
+No sleep-based timing workaround or direct DB status repair is included in this test. Re-run
+all checks after the SDK race is fixed. Aether source was not modified during this investigation.
+
+## 2026-09-11 direct-path follow-up: passed
+
+User selected `directly: true` without modifying Aether. Automatic preparation was enabled;
+`BackgroundJob:WithHostedService=false` was deliberately retained during verification. A fresh
+isolated database (`vNext_autoindex_direct`, domain `autoindexdirect`) preserved the earlier
+failed-run evidence. All **6 checks passed**, and **13 jobs completed with zero retries**.
+The 409 conflict and 500 extra-data cast failure are intentional rollback probes, not unexpected
+failures. Tables are small/empty fixtures; this run is not a populated 10M-row migration benchmark.
+
+This closes the direct-path scenario. Commit-before-schedule crashes and Pending/Retrying
+recovery remain SDK limitations; the earlier failure above has not been erased or fixed.

api-tests/attribute-index-preparation/test_auto_indexes.py

@@ -0,0 +1,142 @@
+#!/usr/bin/env python3
+"""Real-host regression: publication -> durable Aether job -> Dapr -> generated indexes.
+
+Requires an isolated local worktree runtime. SQL is used for assertions and a held
+read lock only; all definitions/data are published through the HTTP API.
+"""
+import json
+import os
+import subprocess
+import time
+import urllib.error
+import urllib.request
+import uuid
+
+BASE = os.environ['VNEXT_BASE_URL'].rstrip('/')
+DOMAIN = os.environ['VNEXT_TEST_DOMAIN']
+PG = os.environ['VNEXT_TEST_PG_CONTAINER']
+DATABASE = os.environ['VNEXT_TEST_DATABASE']
+PREFIX = 'auto-' + uuid.uuid4().hex[:10]
+MASTER = PREFIX + '-master'
+LATEST = PREFIX + '-latest'
+PINNED = PREFIX + '-pinned'
+
+
+def sql(query):
+    return subprocess.check_output(['docker', 'exec', PG, 'psql', '-U', 'postgres', '-d', DATABASE,
+                                    '-At', '-v', 'ON_ERROR_STOP=1', '-c', query], text=True).strip()
+
+
+def publish(kind, key, version, attributes, data=None, allow_failure=False):
+    body = dict(domain=DOMAIN, flow=kind, key=key, version=version, flowVersion='1.0.0', tags=[], attributes=attributes)
+    if data is not None:
+        body['data'] = data
+    req = urllib.request.Request(BASE + '/api/v1/definitions/publish', data=json.dumps(body).encode(),
+                                 headers={'Content-Type': 'application/json'})
+    try:
+        with urllib.request.urlopen(req, timeout=60) as response:
+            return response.status
+    except urllib.error.HTTPError as error:
+        if error.code not in (400, 409) and not allow_failure:
+            raise AssertionError(error.read().decode()) from error
+        print('HTTP', error.code, error.read().decode(), flush=True)
+        return error.code
+
+
+def master(indexed, description=''):
+    return {'type': 'workflow', 'schema': {'type': 'object', 'description': description, 'properties': {
+        'amount': {'type': 'number', 'x-indexed': indexed, 'x-sortable': True, 'x-filterOperators': ['gte']},
+        'name': {'type': 'string', 'x-indexed': indexed, 'x-filterOperators': ['contains']}}}}
+
+
+def workflow(version):
+    return {'type': 'F', 'labels': [{'language': 'en-US', 'label': 'Auto index test'}],
+            'schema': {'domain': DOMAIN, 'flow': 'sys-schemas', 'key': MASTER, 'version': version},
+            'startTransition': {'key': 'start', 'target': 'ready', 'versionStrategy': 'Minor',
+                                'labels': [{'language': 'en-US', 'label': 'Start'}]},
+            'states': [{'labels': [{'language': 'en-US', 'label': 'Ready'}], 'key': 'ready', 'stateType': 1, 'subType': 0, 'versionStrategy': 'Minor',
+                        'onEntries': [], 'onExits': [], 'transitions': []}]}
+
+
+def jobs_finished():
+    rows = json.loads(sql('SELECT coalesce(json_agg(t), \'[]\'::json) FROM '
+                         '(SELECT "Status", "RetryCount", "LastError" FROM sys_queues."BackgroundJobs" '
+                         'WHERE "HandlerName" = \'attribute-index.prepare\') t'))
+    assert not any(row['Status'] == 3 for row in rows), rows
+    return rows and all(row['Status'] == 2 for row in rows)
+
+
+def wait(predicate, timeout=45):
+    end = time.monotonic() + timeout
+    while time.monotonic() < end:
+        if predicate():
+            return
+        time.sleep(0.25)
+    raise AssertionError('Timed out waiting for automatic preparation')
+
+
+def projections(flow):
+    return int(sql("SELECT count(*) FROM information_schema.columns WHERE table_schema='" +
+                   flow.replace('-', '_') + "' AND table_name='InstancesData' AND is_generated='ALWAYS'"))
+
+
+def main():
+    v1 = '1.0.0-pkg.1.0.0'
+    v2 = '1.0.0-pkg.1.0.1'
+    assert publish('sys-schemas', MASTER, v1, master(False)) == 200
+    seed = [{'key': 'seed', 'version': '1.0.0', 'attributes': {'amount': 12, 'name': 'İstanbul'}}]
+    assert publish('sys-flows', LATEST, v1, workflow('latest')) == 200
+    assert publish('sys-flows', PINNED, v1, workflow(v1)) == 200
+    wait(jobs_finished)
+    assert projections(LATEST) == projections(PINNED) == 0
+    print('PASS unindexed publication produces no generated columns')
+
+    assert publish('sys-schemas', MASTER, v2, master(True)) == 200
+    wait(jobs_finished)
+    assert projections(LATEST) == 3
+    assert projections(PINNED) == 0
+    physical = LATEST.replace('-', '_')
+    assert int(sql(f'SELECT count(*) FROM "{physical}"."AttributeIndexCatalog" WHERE "Ready"')) == 3
+    assert int(sql(f"SELECT count(*) FROM pg_indexes WHERE schemaname='{physical}' AND indexname LIKE '%_trgm'")) == 1
+    print('PASS Master package update prepares latest reference; full revision stays pinned')
+
+    # Keep a read lock while publishing an equivalent physical definition. Taking ACCESS EXCLUSIVE
+    # on this path would time out; a real no-op finishes without retries while the lock remains held.
+    lock = subprocess.Popen(['docker', 'exec', '-i', PG, 'psql', '-U', 'postgres', '-d', DATABASE,
+                             '-At', '-v', 'ON_ERROR_STOP=1'], stdin=subprocess.PIPE, stdout=subprocess.PIPE,
+                            stderr=subprocess.PIPE, text=True)
+    try:
+        lock.stdin.write(f'BEGIN; LOCK TABLE "{physical}"."InstancesData" IN ACCESS SHARE MODE; SELECT \'locked\';\n')
+        lock.stdin.flush()
+        while lock.stdout.readline().strip() != 'locked':
+            assert lock.poll() is None
+        assert publish('sys-schemas', MASTER, '1.0.0-pkg.1.0.2', master(True, 'description only')) == 200
+        wait(jobs_finished, timeout=15)
+        assert sql('SELECT count(*) FROM sys_queues."BackgroundJobs" WHERE "HandlerName" = \'attribute-index.prepare\' AND "RetryCount" > 0') == '0'
+        print('PASS equivalent publication completes while a conflicting read lock is held (no DDL/retry)')
+    finally:
+        lock.stdin.write('ROLLBACK;\n'); lock.stdin.close(); lock.wait(timeout=10)
+
+    before = sql('SELECT count(*) FROM sys_queues."BackgroundJobs"')
+    assert publish('sys-flows', LATEST, v1, workflow('latest')) == 409
+    assert sql('SELECT count(*) FROM sys_queues."BackgroundJobs"') == before
+    print('PASS conflicting publication creates no job')
+
+    # The extra-data cast rejects a non-component flow AFTER the definition and seed have been
+    # written. Exercise transaction rollback through the real HTTP middleware and two DbContexts.
+    failed_flow = PREFIX + '-rollback'
+    assert publish('sys-flows', failed_flow, v1, workflow('latest'), seed, allow_failure=True) == 500
+    assert sql(f'SELECT count(*) FROM sys_flows."Instances" WHERE "Key" = \'{failed_flow}\'') == '0'
+    assert sql('SELECT count(*) FROM sys_queues."BackgroundJobs"') == before
+    print('PASS exception after definition writes rolls back publication and leaves no job')
+
+    # New workflow revision moving a pinned reference must also prepare without another Master publish.
+    assert publish('sys-flows', PINNED, v2, workflow('latest')) == 200
+    wait(jobs_finished)
+    assert projections(PINNED) == 3
+    print('PASS workflow reference update triggers automatic preparation')
+    print(json.dumps({'result': 'passed', 'checks': 6, 'base_url': BASE, 'domain': DOMAIN, 'prefix': PREFIX}))
+
+
+if __name__ == '__main__':
+    raise SystemExit('Historical automatic experiment was removed. Run test_manual_indexes.py for the current contract.')

api-tests/attribute-index-preparation/test_manual_indexes.py

@@ -0,0 +1,97 @@
+#!/usr/bin/env python3
+"""Publish -> no automatic index work -> offline CLI SQL -> DBA-style execution in an isolated fixture.
+
+Only run against an explicitly authorized isolated local test domain. Generated SQL
+is executed by this test harness, never by the CLI/runtime. Existing flows are untouched.
+"""
+import json
+import os
+from pathlib import Path
+import subprocess
+import tempfile
+import urllib.error
+import urllib.request
+import uuid
+
+BASE = os.environ['VNEXT_BASE_URL'].rstrip('/')
+DOMAIN = os.environ['VNEXT_TEST_DOMAIN']
+PG = os.environ['VNEXT_TEST_PG_CONTAINER']
+DATABASE = os.environ['VNEXT_TEST_DATABASE']
+CLI = Path(os.environ.get('VNEXT_CLI_ROOT', Path(__file__).resolve().parents[3] / 'vnext-workflow-cli'))
+NODE = os.environ.get('NODE_BINARY', 'node')
+PREFIX = 'manual-' + uuid.uuid4().hex[:10]
+MASTER, FLOW = PREFIX + '-master', PREFIX + '-flow'
+
+
+def sql(query):
+    return subprocess.check_output(['docker', 'exec', '-i', PG, 'psql', '-U', 'postgres', '-d', DATABASE,
+                                    '-At', '-v', 'ON_ERROR_STOP=1'], input=query, text=True).strip()
+
+
+def publish(body):
+    request = urllib.request.Request(BASE + '/api/v1/definitions/publish', data=json.dumps(body).encode(),
+                                     headers={'Content-Type': 'application/json'})
+    try:
+        with urllib.request.urlopen(request, timeout=60) as response:
+            assert response.status == 200
+    except urllib.error.HTTPError as error:
+        raise AssertionError(error.read().decode()) from error
+
+
+def component(kind, key, attributes, version):
+    return {'domain': DOMAIN, 'flow': kind, 'flowVersion': '1.0.0', 'key': key, 'version': version,
+            'tags': ['manual-index-test'], 'attributes': attributes}
+
+
+def main():
+    if os.environ.get('VNEXT_ALLOW_ISOLATED_TEST_DDL') != 'true':
+        raise SystemExit('Requires VNEXT_ALLOW_ISOLATED_TEST_DDL=true and an isolated local test domain.')
+    before = sql("SELECT count(*) FROM sys_queues.\"BackgroundJobs\" WHERE \"HandlerName\"='attribute-index.prepare';")
+    version = '1.0.0-pkg.1.0.0'
+    master = component('sys-schemas', MASTER, {'type': 'workflow', 'schema': {'type': 'object', 'properties': {
+        'amount': {'type': 'number', 'x-indexed': True, 'x-filterOperators': ['gte']},
+        'name': {'type': 'string', 'x-indexed': True, 'x-filterOperators': ['contains']}}}}, version)
+    flow = component('sys-flows', FLOW, {
+        'type': 'F', 'labels': [{'language': 'en-US', 'label': 'Manual index test'}],
+        'schema': {'domain': DOMAIN, 'flow': 'sys-schemas', 'key': MASTER, 'version': 'latest'},
+        'startTransition': {'key': 'start', 'target': 'ready', 'versionStrategy': 'Minor',
+                            'labels': [{'language': 'en-US', 'label': 'Start'}]},
+        'states': [{'labels': [{'language': 'en-US', 'label': 'Ready'}], 'key': 'ready', 'stateType': 1,
+                    'subType': 0, 'versionStrategy': 'Minor', 'onEntries': [], 'onExits': [], 'transitions': []}]
+    }, version)
+    publish(master)
+    publish(flow)
+    schema = FLOW.replace('-', '_')
+    assert sql(f"SELECT count(*) FROM information_schema.columns WHERE table_schema='{schema}' AND table_name='InstancesData' AND is_generated='ALWAYS';") == '0'
+    assert sql("SELECT count(*) FROM sys_queues.\"BackgroundJobs\" WHERE \"HandlerName\"='attribute-index.prepare';") == before
+    print('PASS indexed Master/workflow publication creates no index jobs or projections', flush=True)
+    with tempfile.TemporaryDirectory(prefix='vnext-manual-indexes-') as folder:
+        root = Path(folder)
+        (root/'components'/'Schemas').mkdir(parents=True)
+        (root/'components'/'Workflows').mkdir(parents=True)
+        (root/'vnext.config.json').write_text(json.dumps({'domain': DOMAIN, 'paths': {
+            'componentsRoot': 'components', 'workflows': 'Workflows', 'schemas': 'Schemas'}}))
+        (root/'components'/'Schemas'/'master.json').write_text(json.dumps(master))
+        (root/'components'/'Workflows'/'flow.json').write_text(json.dumps(flow))
+        subprocess.run([NODE, str(CLI/'bin'/'workflow.js'), 'indexes', 'generate', '--flow', FLOW,
+                        '--output', 'sql'], cwd=root, check=True)
+        generated = next((root/'sql').glob('*/*.sql')).read_text()
+        assert sql(f"SELECT count(*) FROM information_schema.columns WHERE table_schema='{schema}' AND table_name='InstancesData' AND is_generated='ALWAYS';") == '0'
+        print('PASS CLI only writes SQL; database is unchanged', flush=True)
+        sql(generated)  # Explicit DBA-style execution by test harness, confined to its unique flow.
+        assert sql(f'SELECT count(*) FROM "{schema}"."AttributeIndexCatalog" WHERE "Ready";') == '3'
+        oids = sql(f"SELECT indexrelid FROM pg_index WHERE indrelid='\"{schema}\".\"InstancesData\"'::regclass ORDER BY indexrelid;")
+        sql(generated)
+        assert sql(f"SELECT indexrelid FROM pg_index WHERE indrelid='\"{schema}\".\"InstancesData\"'::regclass ORDER BY indexrelid;") == oids
+        print('PASS DBA execution prepares catalog, replay preserves indexes', flush=True)
+        master['version'] = '1.0.0-pkg.1.0.1'
+        master['attributes']['schema']['properties']['another'] = {'type': 'integer', 'x-indexed': True}
+        publish(master)
+        assert sql(f'SELECT count(*) FROM "{schema}"."AttributeIndexCatalog" WHERE "Ready";') == '3'
+        assert sql("SELECT count(*) FROM sys_queues.\"BackgroundJobs\" WHERE \"HandlerName\"='attribute-index.prepare';") == before
+        print('PASS subsequent Master publication leaves index maintenance to DBA', flush=True)
+    print(json.dumps({'result': 'passed', 'checks': 4, 'base_url': BASE, 'domain': DOMAIN, 'prefix': PREFIX}))
+
+
+if __name__ == '__main__':
+    main()