job_runs: report the run page and the outcome, not bundle run's event

The wait borrowed `bundle run`'s progress event to report a run, which put a
timestamp in the middle of a deploy's output, made dresources depend on the
`bundle run` progress package, and repeated on failure the state message the
error already carries. A deploy now prints the run page and, once the run
succeeds, its outcome; every state change still goes to the log. bundle/run is
untouched again, so the shared JobStateTracker is no longer needed.

Review cleanups along with it:

- ignore_remote_changes says why each field is ignored rather than calling all
  of them immutable: job_id is, the parameter fields come back resolved against
  the job's defaults, and the rest are never returned. The reason is what
  `bundle plan` shows the user.
- Restore the note that `field: ""` matches nothing, which still applies to the
  root recreate_on_changes rule.
- Drop TestLookupReferencePreDeploy_JobRunReferencesAreRaw, left over from the
  CheckSettled iteration: it asserts framework behaviour this branch does not
  change.
- testCRUD reads the resource again after WaitAfterCreate instead of special
  casing job_runs, the way the DoUpdate block below it already does.
- Drop the RecordRequests failed_run already inherits, and comment the read_id.py
  call that exists only to register a replacement.

Author: radakam

acceptance/bundle/resources/job_runs/basic/output.txt

@@ -35,7 +35,7 @@ Resources:
 Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-basic/default/files...
 Deploying resources...
 job run [MY_RUN_ID]: Run URL: [DATABRICKS_URL]/jobs/[MY_JOB_ID]/runs/[MY_RUN_ID]?o=[NUMID]
-job run [MY_RUN_ID]: [TIMESTAMP] "my-job" TERMINATED SUCCESS
+job run [MY_RUN_ID]: SUCCESS
 Updating deployment state...
 Deployment complete!
 

acceptance/bundle/resources/job_runs/failed_run/output.txt

@@ -4,7 +4,6 @@
 Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/[UNIQUE_NAME]/files...
 Deploying resources...
 job run [MY_RUN_ID]: Run URL: [RUN_URL]
-job run [MY_RUN_ID]: [TIMESTAMP] "test-job-[UNIQUE_NAME]" INTERNAL_ERROR FAILED Task main failed with message: Workload failed, see run output for details.
 Error: cannot create resources.job_runs.my_run: waiting after creating id=[MY_RUN_ID]: run did not succeed: FAILED: Task main failed with message: Workload failed, see run output for details.
 task "main": RuntimeError: intentional failure
 run page: [RUN_URL]
@@ -28,7 +27,6 @@ Plan: 2 to add, 0 to change, 1 to delete, 1 unchanged
 Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/[UNIQUE_NAME]/files...
 Deploying resources...
 job run [MY_RUN_ID_2]: Run URL: [RUN_URL]
-job run [MY_RUN_ID_2]: [TIMESTAMP] "test-job-[UNIQUE_NAME]" INTERNAL_ERROR FAILED Task main failed with message: Workload failed, see run output for details.
 Error: cannot recreate resources.job_runs.my_run: waiting after creating id=[MY_RUN_ID_2]: run did not succeed: FAILED: Task main failed with message: Workload failed, see run output for details.
 task "main": RuntimeError: intentional failure
 run page: [RUN_URL]

acceptance/bundle/resources/job_runs/failed_run/script

@@ -18,6 +18,7 @@ title "the failed run is recorded, and not having succeeded is drift"
 trace read_id.py my_run
 trace $CLI bundle plan
 
+# Name the job too, so the run-now bodies below say which number it is.
 read_id.py my_job > /dev/null
 
 # The recreate re-runs the job instead of accepting the recorded failure.

acceptance/bundle/resources/job_runs/failed_run/test.toml

@@ -1,7 +1,6 @@
 # job_runs is a direct-engine-only resource; the Terraform provider has no
 # equivalent, so restrict the matrix to direct.
 EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]
-RecordRequests = true
 
 # Runs the failing job for real, so the message the deploy names the task with is
 # one a workspace reported rather than one the test server wrote. Serverless needs

acceptance/bundle/resources/job_runs/job_parameters/output.txt

@@ -4,7 +4,7 @@
 Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-job-parameters/default/files...
 Deploying resources...
 job run [MY_RUN_ID]: Run URL: [DATABRICKS_URL]/jobs/[MY_JOB_ID]/runs/[MY_RUN_ID]?o=[NUMID]
-job run [MY_RUN_ID]: [TIMESTAMP] "my-job" TERMINATED SUCCESS
+job run [MY_RUN_ID]: SUCCESS
 Updating deployment state...
 Deployment complete!
 

acceptance/bundle/resources/job_runs/redeploy/output.txt

@@ -4,7 +4,7 @@
 Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-redeploy/default/files...
 Deploying resources...
 job run [MY_RUN_ID]: Run URL: [DATABRICKS_URL]/jobs/[MY_JOB_ID]/runs/[MY_RUN_ID]?o=[NUMID]
-job run [MY_RUN_ID]: [TIMESTAMP] "my-job" TERMINATED SUCCESS
+job run [MY_RUN_ID]: SUCCESS
 Updating deployment state...
 Deployment complete!
 
@@ -90,7 +90,7 @@ Resources:
 Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/job-runs-redeploy/default/files...
 Deploying resources...
 job run [MY_RUN_ID_2]: Run URL: [DATABRICKS_URL]/jobs/[MY_JOB_ID]/runs/[MY_RUN_ID_2]?o=[NUMID]
-job run [MY_RUN_ID_2]: [TIMESTAMP] "my-job" TERMINATED SUCCESS
+job run [MY_RUN_ID_2]: SUCCESS
 Updating deployment state...
 Deployment complete!
 

acceptance/bundle/resources/job_runs/wait/output.txt

@@ -4,7 +4,7 @@
 Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/[UNIQUE_NAME]/files...
 Deploying resources...
 job run [MY_RUN_ID]: Run URL: [RUN_URL]
-job run [MY_RUN_ID]: [TIMESTAMP] "test-job-[UNIQUE_NAME]" TERMINATED SUCCESS
+job run [MY_RUN_ID]: SUCCESS
 Updating deployment state...
 Deployment complete!
 

bundle/direct/bundle_plan_test.go

@@ -308,41 +308,6 @@ func TestLookupReferencePreDeploy_FinishedJobRun(t *testing.T) {
 	assert.Equal(t, jobs.RunResultStateSuccess, value)
 }
 
-// A resource reads the run the workspace reports. These paths are absent from
-// JobRunState, so they resolve from the remote, and the failure-shaped values
-// below come back unrewritten. result_state is the exception: it resolves to the
-// outcome PrepareState requires, which TestJobRunOutcomeIsDrift keeps from
-// standing in for one the run never reached.
-func TestLookupReferencePreDeploy_JobRunReferencesAreRaw(t *testing.T) {
-	b := bundleWithSkippedJobRun(t, &dresources.JobRunRemote{
-		RunId:       123,
-		RunName:     "my-job",
-		RunPageUrl:  "https://myworkspace.databricks.test/jobs/456/runs/123",
-		ResultState: jobs.RunResultStateFailed,
-		State: &jobs.RunState{
-			LifeCycleState: jobs.RunLifeCycleStateTerminated,
-			ResultState:    jobs.RunResultStateFailed,
-			StateMessage:   "task main failed",
-		},
-	})
-
-	for field, want := range map[string]any{
-		"state.result_state":     jobs.RunResultStateFailed,
-		"state.life_cycle_state": jobs.RunLifeCycleStateTerminated,
-		"state.state_message":    "task main failed",
-		"run_id":                 int64(123),
-		"run_name":               "my-job",
-		"run_page_url":           "https://myworkspace.databricks.test/jobs/456/runs/123",
-	} {
-		t.Run(field, func(t *testing.T) {
-			value, err := b.LookupReferencePreDeploy(t.Context(), structpath.MustParsePath(jobRunKey+"."+field))
-
-			require.NoError(t, err)
-			assert.Equal(t, want, value)
-		})
-	}
-}
-
 // References are served from the remote state cache only for a run the plan
 // skips, and any outcome other than the required SUCCESS is a recreate.
 func TestJobRunOutcomeIsDrift(t *testing.T) {

bundle/direct/dresources/all_test.go

@@ -1000,19 +1000,18 @@ func testCRUD(t *testing.T, group string, adapter *Adapter, client *databricks.W
 			"unexpected differences between remappedState and remappedRemoteStateFromCreate")
 	}
 
-	// Only a job run's state moves while WaitAfterCreate blocks, and its result_state
-	// fills in only then, so its field checks below need the settled state.
-	waitSettlesState := group == "job_runs"
-
 	remoteStateFromWaitCreate, err := adapter.WaitAfterCreate(ctx, createdID, newState)
 	require.NoError(t, err)
 	if remoteStateFromWaitCreate != nil {
-		if waitSettlesState {
-			remappedState, err = adapter.RemapState(remoteStateFromWaitCreate)
-			require.NoError(t, err)
-		} else {
-			require.Equal(t, remote, remoteStateFromWaitCreate)
-		}
+		// The resource can move while the wait blocks (a job run only settles
+		// then), so the field checks below use what the wait reports. It must
+		// match a read taken right after it.
+		remotePostWait, err := adapter.DoRead(ctx, createdID)
+		require.NoError(t, err)
+		require.Equal(t, remotePostWait, remoteStateFromWaitCreate)
+
+		remappedState, err = adapter.RemapState(remoteStateFromWaitCreate)
+		require.NoError(t, err)
 	}
 
 	if adapter.HasDoUpdate() {

bundle/direct/dresources/job_run.go

@@ -10,7 +10,6 @@ import (
 	"time"
 
 	"github.com/databricks/cli/bundle/config/resources"
-	"github.com/databricks/cli/bundle/run/progress"
 	"github.com/databricks/cli/libs/cmdio"
 	"github.com/databricks/cli/libs/log"
 	"github.com/databricks/cli/libs/workspaceurls"
@@ -168,10 +167,13 @@ func (r *ResourceJobRun) WaitAfterCreate(ctx context.Context, id string, _ *JobR
 		return nil, err
 	}
 
-	// A run can take hours, so report progress like `bundle run` does. pageURL
+	// A run can take hours, so report the run page as soon as it is known. pageURL
 	// outlives the poll so an abandoned wait can still link the run.
-	var tracker progress.JobStateTracker
 	var pageURL string
+	// Every state the run passes through is logged, but only its outcome is
+	// reported: how many states a run passes through varies with how long its
+	// compute takes to start, and a deploy's output has to stay reproducible.
+	var logged jobs.RunState
 	// Polled here rather than through the SDK waiter, which halts with an error of
 	// its own on the INTERNAL_ERROR a run whose task failed reports, hiding the
 	// task that failed.
@@ -182,8 +184,14 @@ func (r *ResourceJobRun) WaitAfterCreate(ctx context.Context, id string, _ *JobR
 		if err != nil {
 			return nil, retries.Halt(err)
 		}
-		pageURL = cmp.Or(pageURL, run.RunPageUrl)
-		logRunProgress(ctx, run, &tracker)
+		if pageURL == "" && run.RunPageUrl != "" {
+			pageURL = run.RunPageUrl
+			reportRunLine(ctx, runID, "Run URL: "+workspaceurls.ModernizeJobRunPageURL(pageURL))
+		}
+		if run.State.LifeCycleState != logged.LifeCycleState || run.State.ResultState != logged.ResultState {
+			logged = *run.State
+			log.Info(ctx, strings.TrimSpace(fmt.Sprintf("job run %d: %s %s", runID, logged.LifeCycleState, logged.ResultState)))
+		}
 		if !runIsTerminal(run.State.LifeCycleState) {
 			return nil, retries.Continues(run.State.StateMessage)
 		}
@@ -199,8 +207,10 @@ func (r *ResourceJobRun) WaitAfterCreate(ctx context.Context, id string, _ *JobR
 		return nil, fmt.Errorf("%w%s", err, runPageLine(pageURL))
 	}
 	if run.State.ResultState != jobs.RunResultStateSuccess {
+		// The error names the outcome, so it is not reported twice.
 		return nil, r.runFailedError(ctx, run)
 	}
+	reportRunLine(ctx, runID, string(run.State.ResultState))
 	return makeJobRunRemote(run), nil
 }
 
@@ -284,26 +294,6 @@ func runPageLine(rawURL string) string {
 	return "\nrun page: " + workspaceurls.ModernizeJobRunPageURL(rawURL)
 }
 
-// logRunProgress logs every state change like `bundle run` does, but reports only
-// the run page URL and the final state to the user: how many states a run passes
-// through varies with how long its compute takes to start, and a deploy's output
-// has to stay reproducible.
-func logRunProgress(ctx context.Context, run *jobs.Run, tracker *progress.JobStateTracker) {
-	event, first := tracker.Poll(run)
-	if event == nil {
-		return
-	}
-	log.Info(ctx, event.String())
-	if first && run.RunPageUrl != "" {
-		line := "Run URL: " + workspaceurls.ModernizeJobRunPageURL(run.RunPageUrl)
-		log.Info(ctx, line)
-		reportRunLine(ctx, run.RunId, line)
-	}
-	if runIsTerminal(run.State.LifeCycleState) {
-		reportRunLine(ctx, run.RunId, event.String())
-	}
-}
-
 // runIsTerminal reports whether a run has stopped, whatever it stopped as.
 func runIsTerminal(state jobs.RunLifeCycleState) bool {
 	return state == jobs.RunLifeCycleStateTerminated ||