vrr/qwen-code
2
0
Fork 0
mirror of https://github.com/QwenLM/qwen-code.git synced 2026-10-19 12:43:39 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 16

fix(managed-agent): numeric parity and family-list hoisting for publication families, with end-to-end admission and wedge witnesses (#13815)

Browse source 
* fix(managed-agent): admit background Shell and Monitor capture publication families

The capture-publication reserve refused every payload but foreground Shell:
the contract hard-refused is_background and any toolName but
run_shell_command (#13533 A1), the store's checkpoint gate re-spelled the
same shell-only assumption one layer below, and the hosted turn's
checkpoint fork recorded the canonical-input digest only for shell. The
three gates now share one family list (PUBLISHABLE_TOOL_NAMES) covering
foreground/background Shell and Monitor, numeric bounds accept
integral-valued numerics in parity with the TypeScript mirror, and the
turn fork covers monitoring. Families admit only via the shared fixture
vectors replayed byte-identically in both languages (22 vectors incl.
boundary and exponent-spelling cases); mutation witness: removing the
background arm reddens both suites. End-to-end, admitted background and
Monitor reserves answer grant OPEN through the full Spring/broker/daemon
chain (probe-gated witness lanes).

* test(managed-agent): background-publication witness ITs with shared probe wiring

Adds the end-to-end witnesses built while reconstructing #13533's findings:
HostedBackgroundPublicationIT pins admitted background/Monitor reserves
answering grant OPEN through the real Spring controller/stores/broker/daemon
chain (strict detached-arm pins; environment-declined not_started named as
the unlanded B1 dependency), and HostedRecoveryBlockedWedgeIT keeps the A2
wedge probes as regression coverage fed by a driver-relayed reserve 400.
Both boot a shared PublicationProbeWiring extracted under
com.alibaba.qwen.managedprobes (outside the app's component scan), with the
refusal-diagnostic ProbeAdvice registered only in the publication context.
All lanes are gated behind -Dqwen.wedge.probe=true, leaving the CI
hosted-harness-mysql lane unaffected without it.

* docs(design): managed tool publication background arm (EN/zh-CN)

Design record for the #13533 A1 slice: why the single contract gains a
background/Monitor arm rather than background captures routing through a
second reserve (the digest-exact binding is family-independent, so option B
duplicates the grant lifecycle for no invariant), the admitted payload
families with integral-valued numeric bounds, the three closed shell-only
layers the witness pinned (contract, store checkpoint gate, turn fork), the
scope boundaries (no domain enablement; B1/B2/A2 named separately; the
environment-declined 409 cause recorded as an A3-family follow-up), and the
acceptance criteria with their mutation witnesses.

* docs(design): drop the superseded background-arm drafts

Upstream #13789 landed its own shape of the A1 fix (per-family arms with
directory and its own verification), so the drafts written around this
branch's earlier approach (shared constant, strict markers, no directory)
would record decisions that contradict what main now runs. The remaining
delta's rationale moves to the PR thread.

* test(integration): enumerate the wedge driver in the relay-caller witness

The sdk-java lane pins the exact set of drivers calling the shared relay
helper; adding a new legitimate caller (the recovery-blocked wedge driver)
in the same slice must update the enumeration or every later run of
'Check Hosted proxy header relay' fails 5-vs-6. The witness also keeps its
purpose: a driver hand-rolling its own relay stays absent from the list
and is caught — the new entry only records that this one uses the shared
helper, which is exactly the posture the pin exists to keep.

* test(managed-agent): rename probe witnesses off the Hosted*IT roster fingerprint

The hosted fault-gates lane requires every Hosted*IT source class to run
at least one test; the two probe-gated witnesses added with the A1 slice
always record zero executions without -Dqwen.wedge.probe=true (1 run,
1 skipped, which counts as zero against the roster), so the rerun of the
lane fails at 'Check that every Hosted integration test class ran' with
'::error: ... ran no test in this job'. Make the probe intent structural
by renaming them to one-shot ...Gate classes (the O4Gates' shape): the
failsafe include glob '**/Hosted*IT.java' no longer schedules them, and
both roster family predicates (hosted ^Hosted.*IT$ and non-hosted & non-IT)
miss them. Probe lanes keep invoking by simple name; behavior unchanged.
Verified: test-compile + checkstyle green, no residual old-name references
outside archive snapshots.

* feat(cli): add a test-only H3 admission switch for hosted probe witnesses

QWEN_MANAGED_SESSION_TEST_DOMAINS admits the listed child_run/monitor_run
domains at the two turn-admission checks only, so the probe witnesses can
run against an unmodified bundle while the store-side commit gate keeps
refusing what the frozen registry disables; production never sets it.
Also reverts the publication inputDigest fork to upstream's form: since
the merge, the store-side checkpoint gate admits the monitor family
through PUBLISHABLE_TOOL_NAMES, which made the extra monitoring disjunct
a provable no-op.

* test(managed-agent): harden the hosted probe witnesses and collect them in a profile

- Move ProbeAdvice out of the scanned package into
  com.alibaba.qwen.managedprobes so only a witness that lists it as an
  explicit application source registers it; the wedge witness omits it and
  keeps observing the production envelope as-is.
- Add the hosted-harness-gates failsafe profile collecting
  **/Hosted*Gate.java with the probe flag on, keeping the Hosted*IT
  roster, its CI lane and both roster checks untouched.
- Give both gates the driver-process lifecycle finally, budgets that
  cover the worst arm (driver waitFor 1500s / wedge watch 900s under
  @Timeout 1800s and a 2400s fork ceiling), an accurate deadline-expiry
  assertion message, quoted wrapper paths that reject quotes and
  newlines, per-gate evidence subtrees, and a session sweep that binds
  every probe session id with capped-sample naming.
- Point the witness javadoc at #13533 and PR #13815 instead of the
  removed design doc, and name this PR's actual delta for the monitor
  arm (PUBLISHABLE_TOOL_NAMES extraction and the boundedIntField
  numeric widening) rather than a chain lifted elsewhere.
- Driver: the header now says the reserve refusal is relay-injected and
  never reaches the server; the daemon gets the admission switch; each
  family replays its captured OPEN reserve with the same publicationId
  and only its digests forged, fired before the blocked settlement, and
  the 400 refusal reason must name the digest.

* test(core): pin fractional numeric spellings in the publication payload corpus

Adds shell-timeout-fractional, monitor-idle-timeout-fractional and
monitor-max-events-fractional (each carrying a 1.5 field, all invalid)
so the integral-only clause of boundedIntField is pinned in both
languages by the shared, digest-stamped fixture vectors.

* test(cli): register the test-only admission switch in the process.env guard

Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>

* test(managed-agent): align the witnesses with the H3-enabled registry

Upstream #13889 (#13856) enabled everything the witnesses' simulated
switch was papering over (shell kind and monitor_run domain are now
enabled for submission), so the switch remnants are removed: the driver's
daemon env entry, the process.env guard allowance, and both gates'
javadoc mentions. The gates now assert straight on the plain registry
against the unmodified bundle — exactly the reachability proof the review
demanded for the record-commit path.

The Arm-3 oracle is repointed at the layer that actually answers: a
forged-digest replay of the captured OPEN reserve is refused by
ToolPublicationStore.lockOriginal() with "Broker execution identity
conflicts" before any payload validation, so the assertion checks
status 400 plus the identity-conflict text instead of a digest mention
that can never occur.

* test(managed-agent): pin the arm-3 oracle to the placement/replay refusal layer

Since the upstream merge, ToolPublicationStore.applyLocked() runs
lockPublicationPlacement before lockOriginal, so a forged-digest replay
of the captured OPEN reserve is refused at the placement check —
"Reservation replay conflicts", comparing the replayed binding digest
against the persisted row — before any identity or payload validation.
The wire oracle is repointed to that layer (status 400 plus the
reservation-replay-conflict message); the driver comment, the gate
javadoc and the reviewer plan wording follow the layer that actually
answers.

* test(managed-agent): make the probe-turn verdict terminal instead of wall-clock

driveProbeTurn's completed path latched from an initial 'completed' with
no terminal-state check: after the two threshold exits, the fall-through
waited for !hasActivePrompt but never verified the turn ended as a
completion — a store request that hung got the daemon to fail the turn,
and the driver journaled HOSTED_WEDGE_ABSENT green beside the wedged
probe's evidence. The threshold path now returns 'wedged' only when the
turn is still active past its budget; otherwise it breaks the polling
loop and the transcript gate decides. The completed path now requires
exactly one turn_complete for its promptId on both the threshold exit
and the post-deadline fall-through, mirroring the recovery path's own
assertion. A roster-witness pin in hosted-relay-headers.test.ts keeps
that gate pinned inside the driver's completed path; removing it must
turn the relay test red.

* test(managed-agent): widen the wedge wait, wipe evidence per run and correct gate attribution

- The post-clear driver wait rises from 300 s to 600 s: the driver's own
  post-clear budget runs to roughly 510 s (two sequential 240 s windows
  plus a 30 s transcript fetch), so a legitimate A2 recovery inside it
  would see the intended AssertionError('A2 REGRESSION: ...') route
  through an opaque driver timeout, with Files.copy(driver-final.log)
  skipped by the finally that also destroys the driver.
- Both admission gates wipe their own evidence subtree right after
  Files.createDirectories: runs share the fixed
  target/wedge-evidence directory, so an earlier wedged run's artifacts
  otherwise survive into a later green run's bundle, and node-reports/
  accumulates its SIGUSR2 dumps across runs forever.
- HostedBackgroundPublicationGate's javadoc no longer credits this PR
  with the H3 family admission: #13789 landed that at the merge base,
  and requireCheckpoint already allowed the monitor family there; this
  witness's own delta is PUBLISHABLE_TOOL_NAMES, boundedIntField and
  the shared payloadVectors corpus. The monitor arm's closing claim
  re-points at the store-side checkpoint gate reading the same constant.

* ci(sdk-java): collect the hosted admission gates with a hosted-gates roster family

The hosted-harness-gates profile from this PR's round-2 merges matched
no automated lane: surefire defaults skip *Gate.java, no workflow
referenced the profile, and check-failsafe-reports had no roster mode
for it — every lane stayed green without ever executing the witnesses.
Wire it the way -Po4-mysql-gates is wired:

- A new 'Verify Hosted admission gates' step in the Hosted process fault
  gates job, after that lane's report upload so its clean cannot discard
  the hosted lane's already-uploaded failsafe reports and fg7-cases
  evidence, with a 50 minute ceiling above the profile's 2400 s shared
  fork budget plus the wrapped compile/surefire/checkstyle work, and a
  matching 'Check that every Hosted admission gate class ran' roster
  step through a new hosted-gates family (/^Hosted.*Gate$/).
- check-failsafe-reports.js registers the family beside the o4 modes
  with its usage string updated; Hosted*Gate classes stay out of both
  existing 'hosted' / 'non-hosted' families.
- scripts/tests/check-failsafe-reports.test.js gains the new family's
  happy/outside-family/ran-no-test pins, and the upper bound in
  scripts/tests/hosted-process-ci.test.js moves from (145, 158, 160) to
  (195, 208, 210) for the added step.

* test(integration): bind the completed-path sentinel to driveProbeTurn's transcript gate

doudouOUC's verified mutation on dbId 4255682183 showed the R3-18 sentinel
stayed green after removing the completed-path transcript gate: the slice
started at the first 'const transcript = await daemon.json(', so with the
gate deleted indexOf slid forward into wedgeEvidence, whose predicates are
byte-identical. Bound the inspected source to driveProbeTurn, ending before
wedgeEvidence, so removing the gate now turns the sentinel red. Verified
both directions at this head: 4/4 pass unmutated; deleting the gate fails
the first toContain on '/transcript?cursor=0&limit=256'; wedgeEvidence's
fallback predicate supply is unreachable after slicing.

* test(sdk-java): refuse evidence wipe roots outside the module target

yiliang114's P3 (dbId 4256822620) on the wedge gate's recursive wipe: the
resolved evidence root comes straight from -Dqwen.wedge.evidence and was
recursively cleared with no containment check, so a local run pointed at a
shared directory would delete unrelated content there. Both Hosted*Gate
classes now assert the normalized evidence path stays under the module
target/ before any recursive wipe, in the same assertion style the file
already uses for its wrapper paths.

Verified both directions on this head: with the property pointed outside
target/, HostedRecoveryWedgeGate and HostedBackgroundPublicationGate fail in
under a second at the new assertion and seeded sentinel files outside
survive; with the default target/wedge-evidence path both gates proceed past
the guard and fail only later at the intended MySQL stage (Druid/Flyway
communications link failure against a dead port), leaving
target/wedge-evidence/{wedge,publication} created.

---------

Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>
Co-authored-by: 易良 <1204183885@qq.com>
This commit is contained in:
Shaojin Wen 2026-10-13 13:12:53 +00:00 • committed by GitHub
parent 643bcaa07e
commit cfb594252e
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
16 changed files with 2524 additions and 29 deletions
Download patch file Download diff file Expand all files Collapse all files

32
.github/workflows/sdk-java.yml vendored
View file

@ -738,13 +738,14 @@ jobs:
hosted-harness-mysql:
name: 'Hosted process fault gates / MySQL 8.4 / Java 21'
runs-on: 'ubuntu-latest'
# The step ceilings (35 + 9×10 + 20 = 145 — the Verify and O4 steps
# sit above their failsafe fork timeouts; the FG7 channel classes
# raised the hosted fork 900→1200 s) plus ~13 minutes for the
# The step ceilings (35 + 9×10 + 20 + 50 = 195 — the Verify, admission
# and O4 steps sit above their failsafe fork timeouts (the FG7 channel
# classes raised the hosted fork 900→1200 s; the admission gates share
# a 2400 s fork for both witness classes)) plus ~13 minutes for the
# uncapped setup steps (Checkout..Install MySQL binaries, the report
# checks) need ~158; 160 keeps a thin margin, and each arm buys its
# checks) need ~208; 210 keeps a thin margin, and each arm buys its
# own 10 instead of sharing an envelope.
timeout-minutes: 160
timeout-minutes: 210
services:
mysql:
image: 'mysql:8.4.6'
@ -875,6 +876,27 @@ jobs:
packages/sdk-java/managed-agent-server/target/fg7-cases/
retention-days: 7
- name: 'Verify Hosted admission gates'
# Its clean would otherwise discard the hosted lane's failsafe
# reports and the fg7-cases evidence the hosts upload above, so it
# runs after the fixture. The hosted-harness-gates profile carries a
# 2400 s failsafe fork budget for both admission witness classes
# (@Timeout(1800) each); the same killed-after-timeout pattern as the
# hosted-harness-mysql verify applies.
timeout-minutes: 50
env:
MAVEN_ARGS: '--settings ${{ runner.temp }}/setup-java-m2/settings.xml --toolchains ${{ runner.temp }}/setup-java-m2/toolchains.xml'
MYSQL_PORT: "${{ job.services.mysql.ports['3306'] }}"
run: >-
mvn --batch-mode --no-transfer-progress -f packages/sdk-java/managed-agent-server/pom.xml
-Phosted-harness-gates
-Dnode.executable="$(command -v node)"
-Dqwen.cli.entry="${GITHUB_WORKSPACE}/dist/cli.js"
-Dmysql.url="jdbc:mysql://127.0.0.1:${MYSQL_PORT}/hosted_harness_test?allowPublicKeyRetrieval=true&useSSL=false"
-Dmysql.user=root -Dmysql.password=hosted-fixture clean verify checkstyle:check
- name: 'Check that every Hosted admission gate class ran'
run: 'node scripts/check-failsafe-reports.js hosted-gates packages/sdk-java/managed-agent-server'
- name: 'Verify O4 filesystem process and capacity gates'
id: 'o4_mysql'
if: "${{ !cancelled() && steps.managed_dependencies.outcome == 'success' }}"

1260
integration-tests/helpers/hosted-recovery-blocked-wedge-driver.ts Normal file
View file

File diff suppressed because it is too large Load diff

26
integration-tests/helpers/hosted-relay-headers.test.ts
View file

@ -123,7 +123,9 @@ describe('Hosted proxy header relay', () => {
// the drivers relative to this file, never process.cwd(). Read source
// text because importing a driver throws — each parses process.argv[2]
// at module load. Positive pin only: a new driver hand-rolling its relay
// is absent from the enumerated list and still passes.
// is absent from the enumerated list and still passes. A new legitimate
// caller of the shared helper joins this list deliberately — the list
// itself is what this pin keeps countable.
const dir = dirname(fileURLToPath(import.meta.url));
const callers = readdirSync(dir)
.filter((name) => /^hosted-.*-driver\.ts$/.test(name))
@ -135,9 +137,31 @@ describe('Hosted proxy header relay', () => {
'hosted-channel-fault-driver.ts',
'hosted-latency-driver.ts',
'hosted-process-crash-driver.ts',
'hosted-recovery-blocked-wedge-driver.ts',
'hosted-shell-output-driver.ts',
'hosted-store-failure-driver.ts',
'hosted-workspace-tool-turn-driver.ts',
]);
});
it('pins the probe driver to certify completed turns by their transcript', () => {
// R3-18 sentinel: classifying a probe turn from wall clock alone lets a
// failed turn journal green beside a wedged one's evidence; the
// transcript gate is the only honest completed-path verdict — removing
// it must turn this pin red.
const source = readFileSync(
join(
dirname(fileURLToPath(import.meta.url)),
'hosted-recovery-blocked-wedge-driver.ts',
),
'utf8',
);
const completedArm = source.slice(
source.indexOf('async function driveProbeTurn('),
source.indexOf('async function wedgeEvidence('),
);
expect(completedArm).toContain('/transcript?cursor=0&limit=256');
expect(completedArm).toContain("event.type === 'turn_complete'");
expect(completedArm).toContain('event.promptId === turn.promptId');
});
});

177
packages/core/src/managed-runtime/contracts/managed-tool-publication-v1.fixtures.json
View file

@ -3,6 +3,183 @@
"payloadJson": "{\"toolName\":\"run_shell_command\",\"input\":{\"command\":\"printf hi\",\"description\":\"capture\"}}",
"canonicalInputJson": "{\"command\":\"printf hi\",\"description\":\"capture\"}",
"argsJson": "{\"harnessSessionId\":\"session-1\",\"runtimeSessionId\":\"runtime-session-1\",\"payloadJson\":\"{\\\"toolName\\\":\\\"run_shell_command\\\",\\\"input\\\":{\\\"command\\\":\\\"printf hi\\\",\\\"description\\\":\\\"capture\\\"}}\"}",
"payloadVectors": [
{
"id": "background-shell",
"valid": true,
"payloadJson": "{\"toolName\":\"run_shell_command\",\"input\":{\"command\":\"sleep 300\",\"description\":\"background capture\",\"is_background\":true}}",
"requestDigest": "sha256:c81849f53abb0083aa249bc60c1c574b7e5de6f129ef15eb132e8aa6c9f29ef9",
"argsDigest": "sha256:2fb0a8ddc0c746a7db94c14203c34be2559405bbe79259742d6a46db938b221e"
},
{
"id": "background-shell-timeout",
"valid": true,
"payloadJson": "{\"toolName\":\"run_shell_command\",\"input\":{\"command\":\"sleep 300\",\"timeout\":600000,\"is_background\":true}}",
"requestDigest": "sha256:d1df5e4c8d8a3e9f326603c4c784b4d74330fcd508b0a33d79e517c58b47729e",
"argsDigest": "sha256:232a22d3e08815783a0b7cd667a15b02dbdf2487527c0aac422c024463b821cf"
},
{
"id": "monitor",
"valid": true,
"payloadJson": "{\"toolName\":\"monitor\",\"input\":{\"command\":\"tail -f app.log\",\"description\":\"watch the log\",\"idle_timeout_ms\":30000,\"is_monitor\":true,\"max_events\":500}}",
"requestDigest": "sha256:eab83c00bb3a1f4571292cfdf6570c9f4a966d635c6435c77bbde92defe8c974",
"argsDigest": "sha256:7027c8316a2731ecebf5f9dd4579437df411cd39646b9efd3d24b8f75aee5797"
},
{
"id": "monitor-minimal",
"valid": true,
"payloadJson": "{\"toolName\":\"monitor\",\"input\":{\"command\":\"tail -f app.log\",\"is_monitor\":true}}",
"requestDigest": "sha256:5405e318a24240a32903fa31360fec5f0f603d09179c91e1a35b976839d19e62",
"argsDigest": "sha256:911c9314170715ce5cc1e042f9d4322a2efca2e6737057444ad584ef2ac6c0b9"
},
{
"id": "background-string-flag",
"valid": false,
"payloadJson": "{\"toolName\":\"run_shell_command\",\"input\":{\"command\":\"sleep 300\",\"is_background\":\"true\"}}",
"requestDigest": "sha256:b5eaa0c0983387c8d4c9dd15f03844f0960c22a04b2dfe56d53eab97b90dbb3b",
"argsDigest": "sha256:e280e95a318be2942314f9189cb77a71a9e843b57387f21926361ef2a555306a"
},
{
"id": "background-explicit-false",
"valid": true,
"payloadJson": "{\"toolName\":\"run_shell_command\",\"input\":{\"command\":\"printf hi\",\"is_background\":false}}",
"requestDigest": "sha256:55096daad2f54b6784c7213c20902325d0ebe4b38002519655eb16e6b6178ff1",
"argsDigest": "sha256:979825f64b90d9df3850567dabc993724aa8734e9b12aed127e6bf05b86bb4f0"
},
{
"id": "shell-unknown-key",
"valid": false,
"payloadJson": "{\"toolName\":\"run_shell_command\",\"input\":{\"command\":\"printf hi\",\"is_background\":true,\"shell\":\"bash\"}}",
"requestDigest": "sha256:44ae05b9679d50370f3cac2eded3ab169b0024d84a37f26036310709a6c9fae9",
"argsDigest": "sha256:3e755585b8580a9210cbe6c6309aa5d759062bead8e33a8739851e280d05df87"
},
{
"id": "background-bad-timeout",
"valid": false,
"payloadJson": "{\"toolName\":\"run_shell_command\",\"input\":{\"command\":\"sleep 300\",\"is_background\":true,\"timeout\":600001}}",
"requestDigest": "sha256:2e361a607f8b004daeb1401a0db7fe6053379ad335048a2a284203f47a1d2b2d",
"argsDigest": "sha256:fe59bea464d33e1d18e55464a0107793e1c57f18d6b9f696af3d677bbbf9b7ea"
},
{
"id": "monitor-no-marker",
"valid": true,
"payloadJson": "{\"toolName\":\"monitor\",\"input\":{\"command\":\"tail -f app.log\"}}",
"requestDigest": "sha256:fb9e80ff565fce70d0b8f6cdc615192714c96f3c71ecfea2adf4bdd1c7a7dfdd",
"argsDigest": "sha256:858bba3697d3750ca35b16a3619c3279195ac60ebfc2f4b63fb2fbc5c06c12d1"
},
{
"id": "monitor-string-marker",
"valid": false,
"payloadJson": "{\"toolName\":\"monitor\",\"input\":{\"command\":\"tail -f app.log\",\"is_monitor\":\"true\"}}",
"requestDigest": "sha256:3dc630efb3b77dc056c54edf13a39f9fa7aac784774ed490febd3b24b0eb437a",
"argsDigest": "sha256:c3f0f49053360bebe684a127e65c0a3cdda1d2d96fc89a08a407a777bb7d0d70"
},
{
"id": "monitor-unknown-key",
"valid": false,
"payloadJson": "{\"toolName\":\"monitor\",\"input\":{\"command\":\"tail -f app.log\",\"is_monitor\":true,\"interval\":5}}",
"requestDigest": "sha256:c45dc7b5fa93c49a0f26dc13fa3f7e4d7107864979f125d7e5dccf0a70c0cf89",
"argsDigest": "sha256:f6d15d253892df32f6cebc3c454138bf884eaa2e2934a67c18ebd30c55fc78b3"
},
{
"id": "monitor-bad-idle-timeout",
"valid": false,
"payloadJson": "{\"toolName\":\"monitor\",\"input\":{\"command\":\"tail -f app.log\",\"is_monitor\":true,\"idle_timeout_ms\":600001}}",
"requestDigest": "sha256:ade6675ff565f468c51cfcd971c842e51983cdcaf9d0431cbf24f54b6cafde83",
"argsDigest": "sha256:a4d93e5076db20289f004061081a0f4ef4044c0641bbfb7f286f2fd42a1d160a"
},
{
"id": "monitor-bad-max-events",
"valid": false,
"payloadJson": "{\"toolName\":\"monitor\",\"input\":{\"command\":\"tail -f app.log\",\"is_monitor\":true,\"max_events\":10001}}",
"requestDigest": "sha256:dffecc5653b8a5adcfa95387dab1c1d8592c73ad49e7115f87a004d1295db44e",
"argsDigest": "sha256:0d02a5cd90187df5923d89b2b701c02f2e2e8b2098cb97de3fdfcc1d7c21db61"
},
{
"id": "monitor-empty-command",
"valid": false,
"payloadJson": "{\"toolName\":\"monitor\",\"input\":{\"command\":\"\",\"is_monitor\":true}}",
"requestDigest": "sha256:b1757c13f27fa16ade541fe813017af3dcad2dbd9007a97c55ae174f926c1db5",
"argsDigest": "sha256:543da28d6df0402acc09285b856f99c014b2051b3b38cd07162276552bad7868"
},
{
"id": "unknown-tool",
"valid": false,
"payloadJson": "{\"toolName\":\"write_file\",\"input\":{\"content\":\"x\"}}",
"requestDigest": "sha256:96b7c954f6c5fe4b2fd2a7a2677366f2ca8c8e36cae21d8a76c637be0851f66d",
"argsDigest": "sha256:ee2b252d1cd491425942090e06507c7337b5279df43af31ab718b1b1b5da8708"
},
{
"id": "monitor-idle-exponent",
"valid": true,
"payloadJson": "{\"toolName\":\"monitor\",\"input\":{\"command\":\"tail -f app.log\",\"idle_timeout_ms\":6e5,\"is_monitor\":true}}",
"requestDigest": "sha256:aa3d6fabd8daec37b4d3eaa2b2e60720bc4655016c3657d0b7d41660d3b2b56b",
"argsDigest": "sha256:3858b8f13deb1dfafb7a208efce5e43303c5e1a80e3eced5a41f0a71a9e01dce"
},
{
"id": "shell-timeout-decimal",
"valid": true,
"payloadJson": "{\"toolName\":\"run_shell_command\",\"input\":{\"command\":\"printf hi\",\"timeout\":600000.0}}",
"requestDigest": "sha256:c256a5b770823a8091e1cf7f76854d405971ca42bf8d90f599f3f40dca436e39",
"argsDigest": "sha256:23aade92a5dd8942a163d01a85ad2af48d04c1fc3b27e6e2aeab4839a6c3d8a1"
},
{
"id": "monitor-max-events-exponent",
"valid": true,
"payloadJson": "{\"toolName\":\"monitor\",\"input\":{\"command\":\"tail -f app.log\",\"is_monitor\":true,\"max_events\":1e4}}",
"requestDigest": "sha256:5bf2c66a92e675c71ef7b70bb730a8a1d24de7a46cc9eab7585117e619d166ed",
"argsDigest": "sha256:033f3e2eecb32e7937ab202290fb19d236de3642c94b9b160d4d18f535f5808f"
},
{
"id": "monitor-idle-min-boundary",
"valid": true,
"payloadJson": "{\"toolName\":\"monitor\",\"input\":{\"command\":\"tail -f app.log\",\"idle_timeout_ms\":1,\"is_monitor\":true}}",
"requestDigest": "sha256:3b42f774023e574631b3f358122d6221b750e2b0c5795d0ad4eca2134df8add1",
"argsDigest": "sha256:fa1fa67d24670c6fdd987ef9b3c13dd02642edcf38485f1c57b56b485fa0ed09"
},
{
"id": "monitor-idle-boundary",
"valid": true,
"payloadJson": "{\"toolName\":\"monitor\",\"input\":{\"command\":\"tail -f app.log\",\"idle_timeout_ms\":600000,\"is_monitor\":true}}",
"requestDigest": "sha256:3cffe643746f9ad88d68240d5ae4936cf9c7b2a8bbc933ae2a505e29b3e27fe3",
"argsDigest": "sha256:3858b8f13deb1dfafb7a208efce5e43303c5e1a80e3eced5a41f0a71a9e01dce"
},
{
"id": "monitor-max-events-min-boundary",
"valid": true,
"payloadJson": "{\"toolName\":\"monitor\",\"input\":{\"command\":\"tail -f app.log\",\"is_monitor\":true,\"max_events\":1}}",
"requestDigest": "sha256:7ecef2491369c33dbb963c8973bc1b7b6489ec8e4b20fc90a7bb1bcdea13b45a",
"argsDigest": "sha256:709ade433a39a2588d40a43e573492ca611e327616648bb2c2e0b2ebf4ce3c82"
},
{
"id": "monitor-max-events-boundary",
"valid": true,
"payloadJson": "{\"toolName\":\"monitor\",\"input\":{\"command\":\"tail -f app.log\",\"is_monitor\":true,\"max_events\":10000}}",
"requestDigest": "sha256:ea6855bf1318a3b80dbe18c16111b07971a67d664675c23b9adcc01aa5a94ca5",
"argsDigest": "sha256:033f3e2eecb32e7937ab202290fb19d236de3642c94b9b160d4d18f535f5808f"
},
{
"id": "shell-timeout-fractional",
"valid": false,
"payloadJson": "{\"toolName\":\"run_shell_command\",\"input\":{\"command\":\"printf hi\",\"timeout\":1.5}}",
"requestDigest": "sha256:cb62bf9066976e383e02a4acd6ed95c39ebc0f9a3826ae7297ce919e2e28ffe0",
"argsDigest": "sha256:ffe0129018abbd27827abf67631660619361e68d19a73d79dbfb517bacfb2d09"
},
{
"id": "monitor-idle-timeout-fractional",
"valid": false,
"payloadJson": "{\"toolName\":\"monitor\",\"input\":{\"command\":\"tail -f app.log\",\"idle_timeout_ms\":1.5,\"is_monitor\":true}}",
"requestDigest": "sha256:671f6b007d5c24a526d6510336b6c2e1d8fdeabb396aa37f76a882d74d4ecfe6",
"argsDigest": "sha256:54e1edb1cb6af26ec6f8b0a290db70321871bae616ca5a5e20ae59eba678c513"
},
{
"id": "monitor-max-events-fractional",
"valid": false,
"payloadJson": "{\"toolName\":\"monitor\",\"input\":{\"command\":\"tail -f app.log\",\"is_monitor\":true,\"max_events\":1.5}}",
"requestDigest": "sha256:ffdb53168a65e0e0d5835317178e5804f11cb4bb218767743fe8d18a23247de2",
"argsDigest": "sha256:86af41b3a6163755b9c0ddf30077ec33fd82a2375eacd87b909dcddd3c9d7a88"
}
],
"bindingDigest": "0e9e44f23047cf18a9347b86e3fcd05e2d58748530cac6b044fc390341f78fe8",
"cases": [
{

24
packages/core/src/managed-runtime/managed-tool-publication.test.ts
View file

@ -36,6 +36,13 @@ const fixtures = JSON.parse(
requestDigest: string;
argsDigest: string;
}>;
payloadVectors: Array<{
id: string;
valid: boolean;
payloadJson: string;
requestDigest: string;
argsDigest: string;
}>;
digestVectors: Array<{ id: string; binding: unknown; digest: string }>;
tokenVector: { token: string; hash: string };
bindingDigest: string;
@ -88,6 +95,23 @@ describe('managed-tool-publication/1', () => {
).toBe(fixtures.tokenVector.hash);
});
describe('publication payload families', () => {
const binding = parseToolPublicationBinding(fixtures.cases[0].value);
for (const vector of fixtures.payloadVectors) {
it(`${vector.id}`, () => {
const bound = {
...binding,
requestDigest: vector.requestDigest,
reference: { ...binding.reference, argsDigest: vector.argsDigest },
};
const check = () =>
assertToolPublicationPayload(bound, vector.payloadJson);
if (vector.valid) expect(check).not.toThrow();
else expect(check).toThrow();
});
}
});
it('keeps original Runtime identity and both digests separate from model pairing', () => {
const binding = parseToolPublicationBinding(fixtures.cases[0].value);
expect(binding.modelCallId).not.toBe(binding.reference.callId);

17
packages/sdk-java/managed-agent-server/pom.xml
View file

@ -258,6 +258,23 @@
</plugins>
</build>
</profile>
<profile>
<id>hosted-harness-gates</id>
<build><plugins><plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-failsafe-plugin</artifactId>
<version>${maven-failsafe-plugin.version}</version>
<configuration>
<includes><include>**/Hosted*Gate.java</include></includes>
<failIfNoTests>true</failIfNoTests>
<!-- Probe witnesses stay off the Hosted*IT roster; this
profile collects them and turns the probe flag on. -->
<systemPropertyVariables><qwen.wedge.probe>true</qwen.wedge.probe></systemPropertyVariables>
<forkedProcessTimeoutInSeconds>2400</forkedProcessTimeoutInSeconds>
</configuration>
<executions><execution><goals><goal>integration-test</goal><goal>verify</goal></goals></execution></executions>
</plugin></plugins></build>
</profile>
<profile>
<id>hosted-process-crashes</id>
<build><plugins><plugin>

50
packages/sdk-java/managed-agent-server/src/main/java/com/alibaba/qwen/code/managedagent/store/ToolPublicationContract.java
View file

@ -25,6 +25,12 @@ import java.util.Set;
public final class ToolPublicationContract {
public static final String PROTOCOL = "managed-tool-publication/1";
/** The tool families a capture publication may bind to. Every reserve-path
* gate must take the family list from here — a second hard-coded copy is
* how the Monitor reserve came to be refused one layer below the payload
* contract (#13533, A1). */
public static final Set<String> PUBLISHABLE_TOOL_NAMES = Set.of(
"run_shell_command", "monitor");
public static final int MAX_BODY_BYTES = 64 * 1024;
public static final long MAX_CAPTURE_BYTES = 1L << 41;
public static final long PRODUCER_BYTES = 2_686_976;
@ -155,13 +161,18 @@ public final class ToolPublicationContract {
String toolName = text(body, "toolName");
// H3: the detached capture family publishes background Shell and
// Monitor payloads beside the foreground Shell; each family keeps its
// own closed input shape.
// own closed input shape. Admission takes the family list from one
// place — the store's checkpoint gate reads the same constant, and a
// second hard-coded copy is what refused the Monitor reserve one layer
// below this contract (#13533).
require(PUBLISHABLE_TOOL_NAMES.contains(toolName),
"Only Shell and Monitor payloads can publish");
if ("run_shell_command".equals(toolName)) {
requireShellInput(input);
} else if ("monitor".equals(toolName)) {
requireMonitorInput(input);
} else {
require(false, "Only Shell and Monitor payloads can publish");
throw new IllegalArgumentException("Publishable tool lacks a contract arm");
}
require(("sha256:" + sha256(bytes)).equals(text(binding, "requestDigest")),
"Original payload digest conflicts");
@ -176,10 +187,7 @@ public final class ToolPublicationContract {
input.fieldNames().forEachRemaining(name -> require(
Set.of("command", "timeout", "description", "is_background", "directory")
.contains(name), "Shell input field is invalid"));
require(!input.has("timeout") || input.get("timeout").isIntegralNumber()
&& input.get("timeout").canConvertToInt()
&& input.get("timeout").intValue() >= 1 && input.get("timeout").intValue() <= 600_000,
"Shell timeout is invalid");
boundedIntField(input, "timeout", 600_000, "Shell timeout is invalid");
require(!input.has("description") || input.get("description").isTextual(),
"Shell description is invalid");
require(!input.has("is_background") || input.get("is_background").isBoolean(),
@ -199,19 +207,31 @@ public final class ToolPublicationContract {
"Monitor description is invalid");
require(!input.has("is_monitor") || input.get("is_monitor").isBoolean(),
"Monitor marker is invalid");
require(!input.has("max_events") || input.get("max_events").isIntegralNumber()
&& input.get("max_events").canConvertToInt()
&& input.get("max_events").intValue() >= 1
&& input.get("max_events").intValue() <= 10_000, "Monitor max_events is invalid");
require(!input.has("idle_timeout_ms") || input.get("idle_timeout_ms").isIntegralNumber()
&& input.get("idle_timeout_ms").canConvertToInt()
&& input.get("idle_timeout_ms").intValue() >= 1
&& input.get("idle_timeout_ms").intValue() <= 600_000,
"Monitor idle timeout is invalid");
boundedIntField(input, "max_events", 10_000, "Monitor max_events is invalid");
boundedIntField(input, "idle_timeout_ms", 600_000, "Monitor idle timeout is invalid");
require(!input.has("directory") || input.get("directory").isTextual()
&& !input.get("directory").textValue().isEmpty(), "Monitor directory is invalid");
}
/**
* An optional integral-valued numeric field bounded below by 1. Integral
* doubles (6e5, 600000.0) are the same integer the TypeScript mirror
* accepts after JSON.parse normalization — the wire spelling is pinned
* again by the request digest, so accepting the integral form here keeps
* the two implementations' accepted families identical.
*/
private static void boundedIntField(JsonNode input, String key, long maximum, String reason) {
JsonNode value = input.get(key);
if (value == null) {
return;
}
BigDecimal number = value.isNumber() && Double.isFinite(value.doubleValue())
? value.decimalValue() : null;
require(number != null && number.stripTrailingZeros().scale() <= 0
&& number.compareTo(BigDecimal.ONE) >= 0
&& number.compareTo(BigDecimal.valueOf(maximum)) <= 0, reason);
}
public static String bindingDigest(JsonNode binding) {
return sha256(canonicalText(canonical(parse("binding", binding)))
.getBytes(StandardCharsets.UTF_8));

4
packages/sdk-java/managed-agent-server/src/main/java/com/alibaba/qwen/code/managedagent/store/ToolPublicationStore.java
View file

@ -754,8 +754,8 @@ public final class ToolPublicationStore {
item = value;
}
}
require(item != null && ("run_shell_command".equals(text(item, "toolName"))
|| "monitor".equals(text(item, "toolName")))
require(item != null
&& ToolPublicationContract.PUBLISHABLE_TOOL_NAMES.contains(text(item, "toolName"))
&& "in_progress".equals(text(item, "state"))
&& "runtime".equals(text(item, "outcomeSource"))
&& text(b, "modelCallId").equals(text(item, "functionCallId"))

266
packages/sdk-java/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/HostedBackgroundPublicationGate.java Normal file
View file

@ -0,0 +1,266 @@
package com.alibaba.qwen.code.managedagent;
import static org.assertj.core.api.Assertions.assertThat;
import com.alibaba.qwen.code.managedagent.api.WorkspaceSelection;
import com.alibaba.qwen.code.managedagent.service.EmbeddedRuntimeBroker;
import com.alibaba.qwen.code.managedagent.store.ManagedAgentStore;
import com.alibaba.qwen.code.runtimebroker.WorkspaceExecutionProfile;
import com.alibaba.qwen.managedprobes.ProbeAdvice;
import com.alibaba.qwen.managedprobes.PublicationProbeWiring;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.attribute.FileTime;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
import org.junit.jupiter.api.io.TempDir;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.boot.web.servlet.context.ServletWebServerApplicationContext;
import org.springframework.jdbc.core.JdbcTemplate;
/**
* End-to-end witness for #13533 finding A1: the background Shell and
* Monitor payload families were admitted into the publication contract
* earlier, with #13789; this PR extracts the publishable family list into
* {@code ToolPublicationContract.PUBLISHABLE_TOOL_NAMES} and widens the
* integral-valued numerics via {@code boundedIntField}, pinned in both
* languages by the shared {@code payloadVectors} corpus. Admitted
* background {@code run_shell_command} and {@code monitor} calls now
* reach a reserve instead of the removed 400.
*
* <p>Arms (production binaries in-process: MariaDB + embedded broker + real
* hosted-harness daemon running the bundled CLI with the H3 domains enabled
* upstream (#13856); fake model; no cgroup v2 on this host, so the
* ephemeral lane declines every shell-typed start {@code not_started}):
*
* <ul>
* <li>background-reserve: admitted background start reserves and the grant
* answers OPEN (200) — the pre-fix 400 no longer exists. The start is then
* declined {@code not_started} on the ephemeral lane (no cgroup v2 root),
* the grant closes NOT_STARTED (200), and the turn goes recovery-blocked;
* full detached-handle settlement after turn completion remains the unlanded
* B1 arm — pinned and named.
* <li>monitor-reserve: on the merged contract arms, this PR extracted the
* publishable list into
* {@code ToolPublicationContract.PUBLISHABLE_TOOL_NAMES} and widened the
* integral-valued numerics with {@code boundedIntField}; the admitted
* Monitor reserve answers <b>OPEN</b> like the background arm, then
* declines {@code not_started} on the ephemeral lane (no cgroup v2 root),
* closes NOT_STARTED (200), and the turn blocks. The store-side
* checkpoint gate that earlier refused its toolName now reads the same
* constant extracted here.
* <li>refusal-edges-live: per-family raw negatives replayed against the
* captured OPEN grant with only its digests forged — shape-level negatives
* can't be emitted by the turn and the digest-bound protocol admits no
* spoofing; each replay answers 400 whose refusal names the reservation
* replay conflict, since the placement check compares the replayed
* binding digest against the persisted row before identity or payload
* validation; the unit fixture corpus pins the field-shape refusals.
* <li>foreground (v3 lane): reserve OPEN + close NOT_STARTED +
* recovery-blocked — off-Linux the ephemeral lane declines all shell-typed
* executions alike; byte-level foreground regression is held by
* {@code HostedWorkspaceToolTurnIT}'s main workspace/shell lane (G5), run in
* the same verification pass.
* </ul>
*
* <p>Collector-less: run through the {@code hosted-harness-gates} profile
* ({@code mvn -Phosted-harness-gates -Dmysql.url=... -Dnode.executable=...}),
* which includes the {@code Hosted*Gate} test classes and sets the probe
* flag, so the hosted-harness-mysql CI lane and both roster checks stay
* untouched. The explicit source list registers {@link ProbeAdvice} so the
* refusal-diagnosis advice surfaces the contract/store refusal reasons in
* the 400 envelope and the test log (production's handler is deliberately
* generic; the wedge witness observes it as-is).
*/
class HostedBackgroundPublicationGate {
@TempDir
private Path temporary;
@Test
@Timeout(1800)
void backgroundAndMonitorReservesAnswerOpen() throws Exception {
org.junit.jupiter.api.Assumptions.assumeTrue(
"true".equals(System.getProperty("qwen.wedge.probe")),
"background publication witness: skipped unless -Dqwen.wedge.probe=true");
assertThat(System.getProperty("mysql.url")).as("witness requires -Dmysql.url")
.startsWith("jdbc:mysql:");
assertThat(System.getProperty("mysql.user")).as("witness requires -Dmysql.user").isNotBlank();
String node = System.getProperty("node.executable");
assertThat(node).as("Pass -Dnode.executable with an absolute Node.js 22+ path").isNotBlank();
Path cli = Path.of(System.getProperty("qwen.cli.entry", "../../../dist/cli.js"))
.toAbsolutePath().normalize();
assertThat(cli).as("CLI entry -Dqwen.cli.entry=%s", cli).isRegularFile();
Path moduleDir = Path.of(System.getProperty("user.dir")).toAbsolutePath();
Path repoRoot = moduleDir.resolve("../../..").normalize();
assertThat(repoRoot.resolve("integration-tests/helpers/hosted-recovery-blocked-wedge-driver.ts"))
.isRegularFile();
Path evidence = (System.getProperty("qwen.wedge.evidence") != null
? Path.of(System.getProperty("qwen.wedge.evidence")).toAbsolutePath()
: moduleDir.resolve("target/wedge-evidence")).resolve("publication");
Path moduleTarget = moduleDir.resolve("target").normalize();
assertThat(evidence.normalize().startsWith(moduleTarget))
.as("-Dqwen.wedge.evidence must resolve under %s before the recursive wipe", moduleTarget)
.isTrue();
Files.createDirectories(evidence);
// Runs share this fixed evidence directory; a stale wedged run's
// artifacts must not survive into a later green run's bundle, so the
// subtree this invocation owns is wiped before any write.
try (var stale = Files.walk(evidence)) {
stale.sorted(java.util.Comparator.reverseOrder()).forEach(entry -> {
try {
Files.deleteIfExists(entry);
} catch (java.io.IOException ignored) {
}
});
}
Files.createDirectories(evidence);
Path reportsDir = evidence.resolve("node-reports");
Files.createDirectories(reportsDir);
temporary = temporary.toRealPath();
Files.createDirectory(temporary.resolve("runtime"));
String tenant = "publication-" + UUID.randomUUID();
record SessionSpec(String name, String role, String shellCommand) {}
List<SessionSpec> specs = List.of(
new SessionSpec("alpha", "background",
"mkdir -p probe && echo BG_OK > probe/bg.txt"),
new SessionSpec("mu", "monitor", "printf MON_OK\\n; sleep 1"),
new SessionSpec("sigma", "foreground", "echo FG_OK > fg-proof.txt"));
List<Path> workspaces = new ArrayList<>();
for (SessionSpec spec : specs) {
Path dir = Files.createDirectory(temporary.resolve(spec.name()));
Files.createDirectory(dir.resolve("child"));
Files.setLastModifiedTime(dir, FileTime.fromMillis(1));
workspaces.add(dir);
}
Path wrapper = temporary.resolve("node-report-wrapper.sh");
for (String path : List.of(node, reportsDir.toString())) {
assertThat(path).as("wrapper paths must not contain a quote or newline")
.doesNotContain("\"").doesNotContain("\n");
}
Files.writeString(wrapper, "#!/bin/sh\nexec \"" + node
+ "\" --report-on-signal --report-signal=SIGUSR2 --report-directory=\""
+ reportsDir + "\" \"$@\"\n");
wrapper.toFile().setExecutable(true);
var arguments = new ArrayList<>(List.of(
"--server.address=127.0.0.1", "--server.port=0",
"--spring.datasource.url=" + System.getProperty("mysql.url"),
"--spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver",
"--spring.datasource.username=" + System.getProperty("mysql.user"),
"--spring.datasource.password=" + System.getProperty("mysql.password", ""),
"--qwen.managed-agent.session-store.enabled=true",
"--qwen.managed-agent.tool-publication.entry-concurrency=32",
"--qwen.managed-agent.harness.enabled=false",
"--qwen.managed-agent.harness.capability-digest=sha256:" + "a".repeat(64),
"--qwen.managed-agent.runtime-broker.enabled=true",
"--qwen.managed-agent.runtime-broker.port=0",
"--qwen.managed-agent.runtime-broker.token=hosted-tools-broker-token",
"--qwen.managed-agent.runtime-broker.durable-local-process=false",
"--qwen.managed-agent.runtime-broker.trusted-local-reboot-recovery=false",
"--qwen.managed-agent.runtime-broker.workspace-cwd=" + temporary,
"--qwen.managed-agent.runtime-broker.state-directory=" + temporary.resolve("runtime"),
"--qwen.managed-agent.runtime-broker.credential-key-id=test",
"--qwen.managed-agent.runtime-broker.credential-key=AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=",
"--qwen.managed-agent.runtime-broker.node-executable=" + wrapper,
"--qwen.managed-agent.runtime-broker.worker-entry=" + cli,
"--qwen.managed-agent.runtime-broker.verified-workspace-recovery-enabled=false",
"--qwen.managed-agent.runtime-broker.cli-entry=" + cli));
for (int index = 0; index < workspaces.size(); index++) {
String prefix = "--qwen.managed-agent.runtime-broker.workspace-mounts[" + index + "].";
arguments.add(prefix + "tenant-id=" + tenant);
arguments.add(prefix + "storage-id=storage-" + index);
arguments.add(prefix + "root=" + workspaces.get(index));
}
var application = new SpringApplicationBuilder(ManagedAgentServerApplication.class,
PublicationProbeWiring.class, ProbeAdvice.class);
try (var spring = (ServletWebServerApplicationContext) application.run(arguments.toArray(String[]::new))) {
JdbcTemplate jdbc = spring.getBean(JdbcTemplate.class);
var metadata = jdbc.queryForMap("SELECT VERSION() AS version, @@version_comment AS engine");
System.out.println("PUBLICATION_DATABASE " + metadata);
assertThat(metadata.toString().toLowerCase())
.as("witness requires MariaDB/MySQL, got %s", metadata)
.containsAnyOf("mysql", "mariadb");
ManagedAgentStore store = spring.getBean(ManagedAgentStore.class);
var sessions = new ArrayList<Map<String, Object>>();
for (int index = 0; index < specs.size(); index++) {
String workspaceId = "workspace-" + index;
jdbc.update("INSERT INTO managed_workspace_registry (tenant_id, workspace_id,"
+ " workspace_generation, storage_id, display_name, config_ref, policy_ref,"
+ " state) VALUES (?, ?, 1, ?, 'Workspace', ?, ?, 'ACTIVE')",
tenant, workspaceId, "storage-" + index,
WorkspaceExecutionProfile.CONFIG_REF, WorkspaceExecutionProfile.POLICY_REF);
jdbc.update("INSERT INTO managed_workspace_access (tenant_id, workspace_id, actor_id,"
+ " role) VALUES (?, ?, ?, 'OPERATOR')",
tenant, workspaceId, "actor".getBytes(StandardCharsets.UTF_8));
var created = store.insertWorkspaceSessionCommand(tenant, "actor", "create-" + index,
"sha256:" + "a".repeat(64), "qwen-code", null, null, List.of(), null,
new WorkspaceSelection(workspaceId, "child"));
var entry = new LinkedHashMap<String, Object>();
entry.put("sessionId", created.sessionId());
entry.put("workspaceId", workspaceId);
entry.put("directory", workspaces.get(index).resolve("child").toString());
entry.put("toolProfile", "hosted-workspace-shell/1");
entry.put("role", specs.get(index).role());
if (specs.get(index).shellCommand() != null)
entry.put("shellCommand", specs.get(index).shellCommand());
sessions.add(entry);
}
EmbeddedRuntimeBroker broker = spring.getBean(EmbeddedRuntimeBroker.class);
Path phaseFile = temporary.resolve("phase");
Path config = temporary.resolve("driver.json");
var driverConfig = new LinkedHashMap<String, Object>();
driverConfig.put("tenantId", tenant);
driverConfig.put("storeUrl", "http://127.0.0.1:" + spring.getWebServer().getPort());
driverConfig.put("brokerUrl", broker.getBaseUri().toString());
driverConfig.put("cliEntry", cli.toString());
driverConfig.put("arm", "publication");
driverConfig.put("driveFault", true);
driverConfig.put("control", false);
driverConfig.put("phaseFile", phaseFile.toString());
driverConfig.put("clearedFile", temporary.resolve("cleared").toString());
driverConfig.put("wireFile", temporary.resolve("wire.jsonl").toString());
driverConfig.put("evidenceDir", evidence.toString());
driverConfig.put("reportsDir", reportsDir.toString());
driverConfig.put("sessions", sessions);
new ObjectMapper().writeValue(config.toFile(), driverConfig);
Files.writeString(evidence.resolve("sessions.txt"),
sessions.stream().map(s -> s.get("role") + "=" + s.get("sessionId"))
.reduce(tenant, (out, line) -> out + "\n" + line) + "\n");
Path driverLog = temporary.resolve("driver.log");
Process driver = new ProcessBuilder(node, "--import", "tsx",
"integration-tests/helpers/hosted-recovery-blocked-wedge-driver.ts",
config.toString())
.directory(repoRoot.toFile())
.redirectErrorStream(true).redirectOutput(driverLog.toFile()).start();
try {
assertThat(driver.waitFor(1500, TimeUnit.SECONDS))
.as("Driver timeout:\n%s", Files.readString(driverLog)).isTrue();
if (Files.exists(temporary.resolve("wire.jsonl"))) {
Files.copy(temporary.resolve("wire.jsonl"), evidence.resolve("wire.jsonl"),
java.nio.file.StandardCopyOption.REPLACE_EXISTING);
}
Files.copy(driverLog, evidence.resolve("driver-final.log"),
java.nio.file.StandardCopyOption.REPLACE_EXISTING);
String log = Files.readString(driverLog);
System.out.println(log);
assertThat(driver.exitValue()).as("Driver output:\n%s", log).isZero();
assertThat(log).contains("HOSTED_BACKGROUND_PUBLICATION_OK");
} finally {
driver.descendants().forEach(ProcessHandle::destroyForcibly);
if (driver.isAlive()) {
driver.destroyForcibly();
}
}
}
}
}

482
packages/sdk-java/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/HostedRecoveryWedgeGate.java Normal file
View file

@ -0,0 +1,482 @@
package com.alibaba.qwen.code.managedagent;
import static org.assertj.core.api.Assertions.assertThat;
import com.alibaba.qwen.code.managedagent.api.WorkspaceSelection;
import com.alibaba.qwen.code.managedagent.service.EmbeddedRuntimeBroker;
import com.alibaba.qwen.code.managedagent.store.ManagedAgentStore;
import com.alibaba.qwen.code.runtimebroker.WorkspaceExecutionProfile;
import com.alibaba.qwen.managedprobes.PublicationProbeWiring;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.lang.management.ManagementFactory;
import java.lang.management.ThreadInfo;
import java.lang.management.ThreadMXBean;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.attribute.FileTime;
import java.sql.Connection;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
import org.junit.jupiter.api.io.TempDir;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.boot.web.servlet.context.ServletWebServerApplicationContext;
import org.springframework.jdbc.core.JdbcTemplate;
/**
* Issue #13533 finding A2: one recovery-blocked session wedges later turns of
* other sessions on the same hosted harness daemon.
*
* <p>Session α is put into the same recovery-blocked shape the A1 chain left
* (admitted background start → failed reserve → cancel →
* {@code close_not_started} → recovery-blocked turn, session renewing). The
* A1 fix admitted the background Shell/Monitor families into the publication
* contract, so the original 400-chain no longer exists at the endpoint; the
* substitute used here is an injected reserve refusal on α's admitted
* background start, injected once at the driver's relay with the same 400
* {@code invalid_request} wire the pre-fix server answered — every downstream
* step (cancel, close, recovery-block) then runs production code, and the
* leftover shape matches what A1 produced. The substitute is documented in
* the reproduction report for this issue.
*
* <p>Session β (and facultatively γ, or β as a preloaded session, per
* -Dqwen.wedge.preloaded) then runs its turn(s) on the same daemon. The
* regression expectation at this tree is <em>absent of wedge</em>: every
* probe turn completes. If a probe ever wedges again, this test captures the
* JVM thread dump (the managed-agent-server runs in-process), the MariaDB
* transaction/lock state and the session rows, deletes session α's store
* rows, asserts the probe recovers, and then fails as the A2 regression.
*
* <p>Collector-less: run through the {@code hosted-harness-gates} profile
* ({@code mvn -Phosted-harness-gates -Dmysql.url=... -Dnode.executable=...}),
* which includes the {@code Hosted*Gate} test classes and sets the probe
* flag. The H3 domains are enabled upstream (#13856), so no bundle patch
* or test switch is needed. Run with -Dqwen.wedge.control=true for the
* no-fault control run.
*/
class HostedRecoveryWedgeGate {
@TempDir
private Path temporary;
@Test
@Timeout(1800)
void recoveryBlockedSessionMustNotWedgeOtherSessionTurns() throws Exception {
org.junit.jupiter.api.Assumptions.assumeTrue(
"true".equals(System.getProperty("qwen.wedge.probe")),
"A2 reproduction probe: skipped unless -Dqwen.wedge.probe=true");
assertThat(System.getProperty("mysql.url")).as("A2 reproduction requires -Dmysql.url")
.startsWith("jdbc:mysql:");
assertThat(System.getProperty("mysql.user")).as("A2 reproduction requires -Dmysql.user").isNotBlank();
String node = System.getProperty("node.executable");
assertThat(node).as("Pass -Dnode.executable with an absolute Node.js 22+ path").isNotBlank();
Path cli = Path.of(System.getProperty("qwen.cli.entry", "../../../dist/cli.js"))
.toAbsolutePath().normalize();
assertThat(cli).as("CLI entry -Dqwen.cli.entry=%s", cli).isRegularFile();
boolean control = "true".equals(System.getProperty("qwen.wedge.control"));
Path moduleDir = Path.of(System.getProperty("user.dir")).toAbsolutePath();
Path repoRoot = moduleDir.resolve("../../..").normalize();
assertThat(repoRoot.resolve("integration-tests/helpers/hosted-recovery-blocked-wedge-driver.ts"))
.isRegularFile();
Path evidence = (System.getProperty("qwen.wedge.evidence") != null
? Path.of(System.getProperty("qwen.wedge.evidence")).toAbsolutePath()
: moduleDir.resolve("target/wedge-evidence")).resolve("wedge");
Path moduleTarget = moduleDir.resolve("target").normalize();
assertThat(evidence.normalize().startsWith(moduleTarget))
.as("-Dqwen.wedge.evidence must resolve under %s before the recursive wipe", moduleTarget)
.isTrue();
Files.createDirectories(evidence);
// Runs share this fixed evidence directory; a stale wedged run's
// artifacts must not survive into a later green run's bundle, so the
// subtree this invocation owns is wiped before any write.
try (var stale = Files.walk(evidence)) {
stale.sorted(java.util.Comparator.reverseOrder()).forEach(entry -> {
try {
Files.deleteIfExists(entry);
} catch (java.io.IOException ignored) {
}
});
}
Files.createDirectories(evidence);
Path reportsDir = evidence.resolve("node-reports");
Files.createDirectories(reportsDir);
temporary = temporary.toRealPath();
Files.createDirectory(temporary.resolve("runtime"));
String tenant = "wedge-" + UUID.randomUUID();
List<Path> workspaces = new ArrayList<>();
List<String> workspaceIds = List.of("workspace-a", "workspace-b", "workspace-c");
for (String name : List.of("alpha", "beta", "gamma")) {
Path dir = Files.createDirectory(temporary.resolve(name));
Files.createDirectory(dir.resolve("child"));
Files.setLastModifiedTime(dir, FileTime.fromMillis(1));
workspaces.add(dir);
}
Path wrapper = temporary.resolve("node-report-wrapper.sh");
for (String path : List.of(node, reportsDir.toString())) {
assertThat(path).as("wrapper paths must not contain a quote or newline")
.doesNotContain("\"").doesNotContain("\n");
}
Files.writeString(wrapper, "#!/bin/sh\nexec \"" + node
+ "\" --report-on-signal --report-signal=SIGUSR2 --report-directory=\""
+ reportsDir + "\" \"$@\"\n");
wrapper.toFile().setExecutable(true);
var arguments = new ArrayList<>(List.of(
"--server.address=127.0.0.1", "--server.port=0",
"--spring.datasource.url=" + System.getProperty("mysql.url"),
"--spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver",
"--spring.datasource.username=" + System.getProperty("mysql.user"),
"--spring.datasource.password=" + System.getProperty("mysql.password", ""),
"--qwen.managed-agent.session-store.enabled=true",
// The conditional OSS-backed publication stack cannot boot
// outside a provisioned rig, so the real controller and
// stores are registered programmatically below with a stub
// object store (A1's reserve is refused before any object
// call), with only the entry-concurrency property bound.
"--qwen.managed-agent.tool-publication.entry-concurrency=32",
"--qwen.managed-agent.harness.enabled=false",
"--qwen.managed-agent.harness.capability-digest=sha256:" + "a".repeat(64),
"--qwen.managed-agent.runtime-broker.enabled=true",
"--qwen.managed-agent.runtime-broker.port=0",
"--qwen.managed-agent.runtime-broker.token=hosted-tools-broker-token",
"--qwen.managed-agent.runtime-broker.durable-local-process=false",
"--qwen.managed-agent.runtime-broker.trusted-local-reboot-recovery=false",
"--qwen.managed-agent.runtime-broker.workspace-cwd=" + temporary,
"--qwen.managed-agent.runtime-broker.state-directory=" + temporary.resolve("runtime"),
"--qwen.managed-agent.runtime-broker.credential-key-id=test",
"--qwen.managed-agent.runtime-broker.credential-key=AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=",
"--qwen.managed-agent.runtime-broker.node-executable=" + wrapper,
"--qwen.managed-agent.runtime-broker.worker-entry=" + cli,
"--qwen.managed-agent.runtime-broker.verified-workspace-recovery-enabled=false",
"--qwen.managed-agent.runtime-broker.cli-entry=" + cli));
for (int index = 0; index < workspaces.size(); index++) {
String prefix = "--qwen.managed-agent.runtime-broker.workspace-mounts[" + index + "].";
arguments.add(prefix + "tenant-id=" + tenant);
arguments.add(prefix + "storage-id=storage-" + index);
arguments.add(prefix + "root=" + workspaces.get(index));
}
var application = new SpringApplicationBuilder(ManagedAgentServerApplication.class,
PublicationProbeWiring.class);
try (var spring = (ServletWebServerApplicationContext) application.run(arguments.toArray(String[]::new))) {
JdbcTemplate jdbc = spring.getBean(JdbcTemplate.class);
var metadata = jdbc.queryForMap("SELECT VERSION() AS version, @@version_comment AS engine");
System.out.println("A2_DATABASE " + metadata);
assertThat(metadata.toString().toLowerCase())
.as("A2 reproduction requires MariaDB/MySQL, got %s", metadata)
.containsAnyOf("mysql", "mariadb");
ManagedAgentStore store = spring.getBean(ManagedAgentStore.class);
var sessions = new ArrayList<Map<String, Object>>();
String[] profiles = {"hosted-workspace-shell/1", "hosted-workspace-files/1",
"hosted-workspace-files/1"};
for (int index = 0; index < 3; index++) {
String workspaceId = workspaceIds.get(index);
jdbc.update("INSERT INTO managed_workspace_registry (tenant_id, workspace_id,"
+ " workspace_generation, storage_id, display_name, config_ref, policy_ref,"
+ " state) VALUES (?, ?, 1, ?, 'Workspace', ?, ?, 'ACTIVE')",
tenant, workspaceId, "storage-" + index,
WorkspaceExecutionProfile.CONFIG_REF, WorkspaceExecutionProfile.POLICY_REF);
jdbc.update("INSERT INTO managed_workspace_access (tenant_id, workspace_id, actor_id,"
+ " role) VALUES (?, ?, ?, 'OPERATOR')",
tenant, workspaceId, "actor".getBytes(StandardCharsets.UTF_8));
var created = store.insertWorkspaceSessionCommand(tenant, "actor", "create-" + index,
"sha256:" + "a".repeat(64), "qwen-code", null, null, List.of(), null,
new WorkspaceSelection(workspaceId, "child"));
sessions.add(Map.of(
"sessionId", created.sessionId(),
"workspaceId", workspaceId,
"directory", workspaces.get(index).resolve("child").toString(),
"toolProfile", profiles[index],
// α's admitted background start is quick-exit, so the
// leftover execution is cheap to leave unresolved.
"shellCommand", index == 0 ? "sleep 2" : "printf PROBE_OK\\n"));
}
EmbeddedRuntimeBroker broker = spring.getBean(EmbeddedRuntimeBroker.class);
Path phaseFile = temporary.resolve("phase");
Path clearedFile = temporary.resolve("cleared");
Path config = temporary.resolve("driver.json");
var driverConfig = new java.util.LinkedHashMap<String, Object>();
driverConfig.put("tenantId", tenant);
driverConfig.put("storeUrl", "http://127.0.0.1:" + spring.getWebServer().getPort());
driverConfig.put("brokerUrl", broker.getBaseUri().toString());
driverConfig.put("cliEntry", cli.toString());
driverConfig.put("driveFault", !control);
driverConfig.put("control", control);
if (!control) {
driverConfig.put("injectReserveDenyForSessionId",
sessions.get(0).get("sessionId").toString());
}
driverConfig.put("bPreloaded", "true".equals(System.getProperty("qwen.wedge.preloaded")));
driverConfig.put("bCaptureBytes", Long.getLong("qwen.wedge.bCaptureBytes", 0L));
driverConfig.put("bToolProfile", System.getProperty("qwen.wedge.bToolProfile", ""));
driverConfig.put("holdAfterBlockMs", Long.getLong("qwen.wedge.holdMs", 0L));
driverConfig.put("phaseFile", phaseFile.toString());
driverConfig.put("clearedFile", clearedFile.toString());
driverConfig.put("wireFile", temporary.resolve("wire.jsonl").toString());
driverConfig.put("evidenceDir", evidence.toString());
driverConfig.put("reportsDir", reportsDir.toString());
driverConfig.put("sessions", sessions);
new ObjectMapper().writeValue(config.toFile(), driverConfig);
Path driverLog = temporary.resolve("driver.log");
Process driver = new ProcessBuilder(node, "--import", "tsx",
"integration-tests/helpers/hosted-recovery-blocked-wedge-driver.ts",
config.toString())
.directory(repoRoot.toFile())
.redirectErrorStream(true).redirectOutput(driverLog.toFile()).start();
Files.writeString(evidence.resolve("sessions.txt"),
"tenant=" + tenant + "\nalpha=" + sessions.get(0).get("sessionId")
+ "\nbeta=" + sessions.get(1).get("sessionId")
+ "\ngamma=" + sessions.get(2).get("sessionId") + "\n");
try {
if (control) {
assertThat(driver.waitFor(300, TimeUnit.SECONDS))
.as("Control driver timeout:\n%s", Files.readString(driverLog)).isTrue();
Files.copy(driverLog, evidence.resolve("driver-final.log"),
java.nio.file.StandardCopyOption.REPLACE_EXISTING);
String controlLog = Files.readString(driverLog);
System.out.println(controlLog);
assertThat(driver.exitValue())
.as("Control driver output:\n%s", controlLog).isZero();
assertThat(controlLog).contains("HOSTED_WEDGE_CONTROL_OK");
return;
}
boolean wedged = false;
long watchDeadline = System.currentTimeMillis() + 900_000;
while (System.currentTimeMillis() < watchDeadline) {
if (Files.exists(phaseFile)
&& Files.readString(phaseFile).trim().equals("B_WEDGED_CONFIRMED")) {
wedged = true;
break;
}
if (!driver.isAlive()) break;
Thread.sleep(400);
}
Files.copy(driverLog, evidence.resolve("driver.log"),
java.nio.file.StandardCopyOption.REPLACE_EXISTING);
if (!wedged) {
String log = Files.readString(driverLog);
if (!driver.isAlive()) {
assertThat(driver.waitFor(60, TimeUnit.SECONDS))
.as("Driver exit wait:\n%s", log).isTrue();
if (log.contains("HOSTED_WEDGE_ABSENT") && driver.exitValue() == 0) {
// Regression expectation: with α recovery-blocked, the
// probes' turns complete. Anything else is the A2
// wedge returning.
dumpAll(jdbc, evidence, sessions, tenant, "not-wedged");
return;
}
if (log.contains("HOSTED_RECOVERY_WEDGE_OK")) {
dumpAll(jdbc, evidence, sessions, tenant, "wedge-regression");
throw new AssertionError(
"A2 REGRESSION: a probe turn wedged with α recovery-blocked"
+ " and only completed after α's rows were cleared."
+ " See " + evidence + " and driver log:\n" + log);
}
throw new AssertionError("Driver exited unexpectedly (wedged=" + wedged
+ "). Log:\n" + log);
}
throw new AssertionError("Watch deadline expired with the driver still"
+ " running (wedged=" + wedged + "). Log:\n" + log);
}
// --- A2 reproduced: capture the full park-point evidence --------
Files.writeString(evidence.resolve("java-threads.txt"), threadDump());
dumpAll(jdbc, evidence, sessions, tenant, "before-clear");
// --- Clear session α's store rows, like the rig operator did ---
String alpha = sessions.get(0).get("sessionId").toString();
String clearReport = clearSessionRows(jdbc, alpha);
Files.writeString(evidence.resolve("clear-alpha.txt"), clearReport);
Files.writeString(clearedFile, clearReport);
dumpAll(jdbc, evidence, sessions, tenant, "after-clear");
// The driver's post-clear budget is two sequential 240 s
// windows plus a 30 s transcript fetch; exceeding it turns a
// legitimate recovery into an opaque timeout and discards the
// clear-and-recover evidence this witness exists to collect.
assertThat(driver.waitFor(600, TimeUnit.SECONDS))
.as("Driver timeout:\n%s", Files.readString(driverLog)).isTrue();
Files.copy(driverLog, evidence.resolve("driver-final.log"),
java.nio.file.StandardCopyOption.REPLACE_EXISTING);
String log = Files.readString(driverLog);
System.out.println(log);
assertThat(driver.exitValue()).as("Driver output:\n%s", log).isZero();
throw new AssertionError(
"A2 REGRESSION: a probe turn wedged with α recovery-blocked and only"
+ " completed after α's rows were cleared. See " + evidence
+ " and driver log:\n" + log);
} finally {
driver.descendants().forEach(ProcessHandle::destroyForcibly);
if (driver.isAlive()) {
driver.destroyForcibly();
}
}
}
}
private static String threadDump() {
ThreadMXBean bean = ManagementFactory.getThreadMXBean();
StringBuilder out = new StringBuilder();
out.append("threadCount=").append(bean.getThreadCount()).append('\n');
for (ThreadInfo info : bean.dumpAllThreads(true, true)) {
out.append('\n').append('"').append(info.getThreadName()).append('"')
.append(" id=").append(info.getThreadId())
.append(" state=").append(info.getThreadState())
.append(" lock=").append(info.getLockName())
.append(" lockOwner=").append(info.getLockOwnerName())
.append('\n');
for (StackTraceElement frame : info.getStackTrace()) {
out.append(" at ").append(frame).append('\n');
}
for (var monitor : info.getLockedMonitors()) {
out.append(" - locked ").append(monitor).append('\n');
}
for (var ownable : info.getLockedSynchronizers()) {
out.append(" - sync ").append(ownable).append('\n');
}
}
return out.toString();
}
private void dumpAll(JdbcTemplate jdbc, Path evidence, List<Map<String, Object>> sessions,
String tenant, String tag) throws Exception {
StringBuilder out = new StringBuilder();
query(out, jdbc, "PROCESSLIST",
"SELECT ID, USER, HOST, DB, COMMAND, TIME, STATE, LEFT(INFO, 400) AS INFO"
+ " FROM information_schema.PROCESSLIST ORDER BY ID");
query(out, jdbc, "INNODB_TRX",
"SELECT TRX_ID, TRX_STATE, TRX_STARTED, TRX_MYSQL_THREAD_ID, TRX_ROWS_LOCKED,"
+ " TRX_ISOLATION_LEVEL, LEFT(TRX_QUERY, 400) AS TRX_QUERY"
+ " FROM information_schema.INNODB_TRX");
query(out, jdbc, "DATA_LOCKS",
"SELECT ENGINE, ENGINE_LOCK_ID, ENGINE_TRANSACTION_ID, THREAD_ID, OBJECT_SCHEMA,"
+ " OBJECT_NAME, LOCK_TYPE, LOCK_MODE, LOCK_STATUS, LEFT(LOCK_DATA, 120) AS LOCK_DATA"
+ " FROM performance_schema.data_locks");
query(out, jdbc, "INNODB_LOCKS_LEGACY", "SELECT * FROM information_schema.INNODB_LOCKS");
query(out, jdbc, "INNODB_LOCK_WAITS_LEGACY", "SELECT * FROM information_schema.INNODB_LOCK_WAITS");
query(out, jdbc, "METADATA_LOCKS",
"SELECT OBJECT_SCHEMA, OBJECT_NAME, LOCK_TYPE, LOCK_DURATION, LOCK_STATUS,"
+ " SOURCE, OWNER_THREAD_ID FROM performance_schema.metadata_locks");
try {
var status = jdbc.queryForMap("SHOW ENGINE INNODB STATUS");
out.append("=== SHOW ENGINE INNODB STATUS ===\n").append(status.get("Status")).append('\n');
} catch (Exception error) {
out.append("INNODB STATUS failed: ").append(error).append('\n');
}
for (Map<String, Object> session : sessions) {
String id = session.get("sessionId").toString();
query(out, jdbc, "journal_head " + id,
"SELECT tenant_id, session_id, workspace_id, state, recovery_status,"
+ " recovery_detail_code, writer_id, writer_generation,"
+ " journal_revision, committed_sequence, activation_epoch,"
+ " writer_lease_until, updated_at FROM"
+ " qwen_managed_session_journal_head WHERE tenant_id = ? AND"
+ " session_id = ?", tenant, id);
query(out, jdbc, "journal_tx " + id,
"SELECT session_id, operation, command_id, journal_revision, first_sequence,"
+ " last_sequence, created_at FROM qwen_managed_session_journal_tx"
+ " WHERE tenant_id = ? AND session_id = ? ORDER BY journal_revision"
+ " DESC LIMIT 25", tenant, id);
query(out, jdbc, "publications " + id,
"SELECT publication_id, tenant_id, session_id, retention_state, state,"
+ " capture_used_bytes FROM qwen_tool_publication WHERE tenant_id = ?"
+ " AND session_id = ?", tenant, id);
}
dumpSessionRows(jdbc, out, sessions);
Files.writeString(evidence.resolve("db-" + tag + ".txt"), out.toString());
}
private static void query(StringBuilder out, JdbcTemplate jdbc, String label, String sql,
Object... args) {
out.append("=== ").append(label).append(" ===\n");
try {
for (Map<String, Object> row : jdbc.queryForList(sql, args)) {
out.append(row).append('\n');
}
} catch (Exception error) {
out.append("query failed: ").append(error.getMessage()).append('\n');
}
}
/** Every table row keyed by any probe session id, through any session-ish column. */
private static void dumpSessionRows(JdbcTemplate jdbc, StringBuilder out,
List<Map<String, Object>> sessions) throws Exception {
List<String> ids = sessions.stream()
.map(session -> session.get("sessionId").toString()).toList();
String placeholders = String.join(", ", java.util.Collections.nCopies(ids.size(), "?"));
for (String table : sessionTables(jdbc)) {
for (String column : sessionColumns(jdbc, table)) {
out.append("=== rows ").append(table).append('.').append(column).append(" ===\n");
try {
List<Map<String, Object>> rows = jdbc.queryForList(
"SELECT * FROM " + table + " WHERE " + column + " IN ("
+ placeholders + ") LIMIT 40", ids.toArray());
for (Map<String, Object> row : rows) out.append(row).append('\n');
out.append("(count=").append(rows.size());
if (rows.size() == 40) {
out.append(", sample capped at 40");
}
out.append(")\n");
} catch (Exception error) {
out.append("query failed: ").append(error.getMessage()).append('\n');
}
}
}
}
private static List<String> sessionTables(JdbcTemplate jdbc) {
return jdbc.queryForList(
"SELECT DISTINCT table_name FROM information_schema.columns"
+ " WHERE table_schema = DATABASE() AND (column_name = 'session_id'"
+ " OR column_name = 'harness_session_id') ORDER BY table_name",
String.class);
}
private static List<String> sessionColumns(JdbcTemplate jdbc, String table) {
return jdbc.queryForList(
"SELECT column_name FROM information_schema.columns WHERE table_schema = DATABASE()"
+ " AND table_name = ? AND (column_name = 'session_id'"
+ " OR column_name = 'harness_session_id') ORDER BY column_name",
String.class, table);
}
private static String clearSessionRows(JdbcTemplate jdbc, String sessionId) throws Exception {
StringBuilder out = new StringBuilder();
String database;
try (Connection connection = jdbc.getDataSource().getConnection()) {
database = connection.getCatalog();
}
out.append("clearing session ").append(sessionId).append(" in ").append(database).append('\n');
for (int round = 0; round < 8; round++) {
int errors = 0;
out.append("--- round ").append(round).append(" ---\n");
for (String table : sessionTables(jdbc)) {
for (String column : sessionColumns(jdbc, table)) {
try {
int deleted = jdbc.update("DELETE FROM " + table + " WHERE " + column + " = ?",
sessionId);
if (deleted > 0)
out.append("deleted ").append(deleted).append(" from ").append(table)
.append('.').append(column).append('\n');
} catch (Exception error) {
errors++;
out.append("FAILED ").append(table).append('.').append(column).append(' ')
.append(error.getMessage()).append('\n');
}
}
}
if (errors == 0) {
out.append("clear complete in round ").append(round).append('\n');
return out.toString();
}
Thread.sleep(500);
}
out.append("CLEAR INCOMPLETE after 8 rounds\n");
return out.toString();
}
}

26
packages/sdk-java/managed-agent-server/src/test/java/com/alibaba/qwen/code/managedagent/ToolPublicationContractTest.java
View file

@ -91,6 +91,32 @@ class ToolPublicationContractTest {
new byte[65537])).isInstanceOf(IllegalArgumentException.class);
}
/**
* Replays the shared payload vectors — the digests are pinned from the
* TypeScript canonicalizer, so an accepted vector also proves the two
* implementations canonicalize the boolean marker fields byte-identically.
*/
@Test
void acceptsAndRefusesThePublicationPayloadFamilies() throws Exception {
JsonNode suite = JSON.readTree(contracts().resolve(
"managed-tool-publication-v1.fixtures.json").toFile());
JsonNode base = suite.required("cases").get(0).required("value");
for (JsonNode vector : suite.required("payloadVectors")) {
ObjectNode binding = (ObjectNode) base.deepCopy();
binding.put("requestDigest", vector.required("requestDigest").asText());
((ObjectNode) binding.get("reference")).put("argsDigest",
vector.required("argsDigest").asText());
String payload = vector.required("payloadJson").asText();
if (vector.required("valid").asBoolean()) {
ToolPublicationContract.requirePayload(binding, payload);
} else {
assertThatThrownBy(() -> ToolPublicationContract.requirePayload(binding, payload))
.as(vector.required("id").asText())
.isInstanceOf(IllegalArgumentException.class);
}
}
}
@Test
void acceptsEcmascriptCanonicalShellInputWithControlCharacters() throws Exception {
JsonNode suite = JSON.readTree(contracts().resolve(

37
packages/sdk-java/managed-agent-server/src/test/java/com/alibaba/qwen/managedprobes/ProbeAdvice.java Normal file
View file

@ -0,0 +1,37 @@
package com.alibaba.qwen.managedprobes;
import jakarta.servlet.http.HttpServletRequest;
import java.util.Map;
import org.springframework.core.Ordered;
import org.springframework.core.annotation.Order;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.ExceptionHandler;
import org.springframework.web.bind.annotation.RestControllerAdvice;
/**
* Probe-lane diagnosis: the production 400 handler answers with a
* deliberately generic envelope and no log, so contract refusals are
* unreadable on the wire. This advice logs the full cause and embeds it in
* the envelope. It lives outside the scanned {@code com.alibaba.qwen.code}
* package, so only a probe witness that lists it as an explicit application
* source registers it; {@code HostedRecoveryWedgeGate} deliberately omits it
* and observes the production envelope as it is.
*/
@RestControllerAdvice
@Order(Ordered.HIGHEST_PRECEDENCE)
public final class ProbeAdvice {
@ExceptionHandler(IllegalArgumentException.class)
ResponseEntity<Map<String, Object>> probeInvalid(IllegalArgumentException error,
HttpServletRequest request) {
System.out.println("PROBE_CONTRACT_REFUSAL " + request.getMethod() + " "
+ request.getRequestURI() + " query=" + request.getQueryString());
error.printStackTrace(System.out);
return ResponseEntity.status(HttpStatus.BAD_REQUEST).body(Map.of(
"error", Map.of(
"code", "invalid_request",
"message", "The request body is invalid.",
"diagnostic", String.valueOf(error.getMessage()))));
}
}

116
packages/sdk-java/managed-agent-server/src/test/java/com/alibaba/qwen/managedprobes/PublicationProbeWiring.java Normal file
View file

@ -0,0 +1,116 @@
package com.alibaba.qwen.managedprobes;
import com.alibaba.qwen.code.managedagent.api.ToolPublicationController;
import com.alibaba.qwen.code.managedagent.config.ManagedAgentProperties;
import com.alibaba.qwen.code.managedagent.store.ManagedSessionStore;
import com.alibaba.qwen.code.managedagent.store.ToolPublicationAdmissionStore;
import com.alibaba.qwen.code.managedagent.store.ToolPublicationContract;
import com.alibaba.qwen.code.managedagent.store.ToolPublicationDataStore;
import com.alibaba.qwen.code.managedagent.store.ToolPublicationObjectStore;
import com.alibaba.qwen.code.managedagent.store.ToolPublicationStore;
import com.alibaba.qwen.code.runtimebroker.RuntimeBindingRepository;
import com.alibaba.qwen.code.runtimebroker.ToolExecutionRepository;
import java.io.ByteArrayInputStream;
import java.io.InputStream;
import java.time.Duration;
import java.util.concurrent.ConcurrentHashMap;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.transaction.PlatformTransactionManager;
/**
* The real publication controller and grant/data/admission stores with an
* in-memory object store, mirroring the conditional OSS-backed
* {@code ToolPublicationConfiguration} that only provisioned rigs can boot.
* Object bytes are never touched by the probe ITs' reserve/close arms, so
* the store can be a stable in-memory stub.
*
* <p>Shared by {@code HostedBackgroundPublicationGate} and
* {@code HostedRecoveryWedgeGate}. It deliberately contains no request
* advice: the refusal-diagnosis advice (which changes the 400 envelope) must
* exist only in the publication witness's context, since the wedge witness
* observes the production envelope as it is.
*/
@Configuration(proxyBeanMethods = false)
public final class PublicationProbeWiring {
private static final long EXECUTION_BYTES = 268_435_456L;
private static final long SESSION_BYTES = 536_870_912L;
private static final long TENANT_BYTES = 1_073_741_824L;
private static final long ACTIVE_CAPTURES = 8;
@Bean
ToolPublicationObjectStore publicationObjectStore() {
return new InMemoryObjectStore();
}
@Bean
ToolPublicationStore publicationStore(JdbcTemplate jdbc,
PlatformTransactionManager manager, ManagedSessionStore sessions,
ToolExecutionRepository executions, RuntimeBindingRepository bindings) {
return new ToolPublicationStore(jdbc, manager, sessions, executions,
bindings, new ToolPublicationStore.Capacity(EXECUTION_BYTES,
SESSION_BYTES, TENANT_BYTES, ACTIVE_CAPTURES), false);
}
@Bean
ToolPublicationDataStore publicationDataStore(JdbcTemplate jdbc,
PlatformTransactionManager manager, ToolPublicationStore grants,
ManagedSessionStore sessions, ToolPublicationObjectStore objects) {
var operationTimeout = Duration.ofSeconds(60);
var budget = new ToolPublicationDataStore.VerificationBudget(
1_048_576L, Duration.ofMinutes(10));
budget.timeout(operationTimeout, Math.addExact(
EXECUTION_BYTES, ToolPublicationContract.PRODUCER_BYTES));
return new ToolPublicationDataStore(jdbc, manager, grants, sessions,
objects, operationTimeout, Duration.ofSeconds(30), budget);
}
@Bean
ToolPublicationAdmissionStore publicationAdmissionStore(JdbcTemplate jdbc,
PlatformTransactionManager manager, ManagedSessionStore sessions,
ToolPublicationDataStore data) {
return new ToolPublicationAdmissionStore(jdbc, manager, sessions, data);
}
@Bean
ToolPublicationController publicationController(ToolPublicationStore grants,
ToolPublicationDataStore data, ToolPublicationAdmissionStore admissions,
ManagedAgentProperties properties) {
return new ToolPublicationController(grants, data, admissions, properties);
}
static final class InMemoryObjectStore implements ToolPublicationObjectStore {
private final ConcurrentHashMap<String, byte[]> objects = new ConcurrentHashMap<>();
@Override
public void requireUnversioned() {
}
@Override
public void putIfAbsent(String key, byte[] bytes) {
objects.putIfAbsent(key, bytes);
}
@Override
public void deleteIfPresent(String key) {
objects.remove(key);
}
@Override
public InputStream open(String key) {
byte[] bytes = objects.get(key);
if (bytes == null) {
throw new IllegalStateException("Unknown object " + key);
}
return new ByteArrayInputStream(bytes);
}
@Override
public InputStream open(String key, Runnable guard) {
guard.run();
return open(key);
}
}
}

4
scripts/check-failsafe-reports.js
View file

@ -20,17 +20,19 @@ import { existsSync, readdirSync, readFileSync } from 'node:fs';
import path from 'node:path';
const isHosted = (className) => /^Hosted.*IT$/.test(className);
const isGate = (className) => /^Hosted.*Gate$/.test(className);
const [family, ...modules] = process.argv.slice(2);
const optIn = family === 'o4-mysql' || family === 'o4-oss';
const inFamily = new Map([
['hosted', isHosted],
['non-hosted', (className) => /IT$/.test(className) && !isHosted(className)],
['hosted-gates', isGate],
['o4-mysql', (className) => className === 'O4MySqlGate'],
['o4-oss', (className) => className === 'O4OssGate'],
]).get(family);
if (!inFamily || modules.length === 0) {
console.error(
'usage: node scripts/check-failsafe-reports.js hosted|non-hosted|o4-mysql|o4-oss <maven-module-dir>...',
'usage: node scripts/check-failsafe-reports.js hosted|non-hosted|hosted-gates|o4-mysql|o4-oss <maven-module-dir>...',
);
process.exit(2);
}

22
scripts/tests/check-failsafe-reports.test.js
View file

@ -256,6 +256,28 @@ describe('check-failsafe-reports', () => {
expect(check('hosted', noFamily).status).toBe(0);
});
it('hosts admission gates as their own family and rejects drift in both directions', () => {
const gates = module('gates', ['com.example.HostedAdmissionGate'], {
'com.example.HostedAdmissionGate': [1, 0],
});
expect(check('hosted-gates', gates).status).toBe(0);
expect(check('hosted-gates', gates).output).toContain(
'gates: com.example.HostedAdmissionGate ran 1 test(s)',
);
const outside = module('outsiders', ['com.example.HostedFlowIT'], {
'com.example.HostedFlowIT': [1, 0],
});
expect(check('hosted-gates', outside).output).toContain(
'com.example.HostedFlowIT ran outside the hosted-gates family',
);
expect(check('hosted-gates', outside).status).toBe(1);
const unrun = module('unrun-gate', ['com.example.HostedWedgeGate']);
expect(check('hosted-gates', unrun).output).toContain(
'com.example.HostedWedgeGate ran no test in this job',
);
expect(check('hosted-gates', unrun).status).toBe(1);
});
it('refuses an unknown family or a missing module', () => {
expect(check('mariadb', root).status).toBe(2);
expect(check('constructor', root).status).toBe(2);

10
scripts/tests/hosted-process-ci.test.js
View file

@ -262,12 +262,12 @@ describe('Hosted real-process gates', () => {
(total, step) => total + (step['timeout-minutes'] ?? 0),
0,
);
// Step ceilings today (35 + 9x10 + 20); the uncapped setup steps need
// their own allowance — the job comment estimates it at ~13, so the
// enforced floor moves with that estimate, not the older 10.
expect(summed).toBe(145);
// Step ceilings today (35 + 9x10 + 20 + 50); the uncapped setup steps
// need their own allowance — the job comment estimates it at ~13, so
// the enforced floor moves with that estimate, not the older 10.
expect(summed).toBe(195);
expect(job['timeout-minutes']).toBeGreaterThanOrEqual(summed + 13);
expect(job['timeout-minutes']).toBe(160);
expect(job['timeout-minutes']).toBe(210);
});
it.each([