diff --git a/CLAUDE.md b/CLAUDE.md
index 54229f93..35c483ef 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -685,6 +685,13 @@ kbagent component sync-action ACTION_NAME --component-id ID --project ALIAS (--c
 #   sends explicit configData verbatim (skips fetch); branchId omitted from body for production.
 kbagent config examples --component-id ID [--project NAME] [--row]
 kbagent config new --component-id ID [--name NAME] [--project NAME] [--output-dir DIR] [--push --no-files --description D --configuration JSON|@file|- --configuration-file PATH --no-validate --branch ID --dry-run --allow-plaintext-on-encrypt-failure]
+# config new --push --output-dir (vNEXT, #644): the written scaffold now records the created
+#   config's ID (_keboola.config_id, quoted so numeric IDs stay strings) and is placed in the
+#   subtree of the branch the config was ACTUALLY created in (--branch or active branch;
+#   unregistered branches are added to the manifest like `sync pull --branch` would). Before,
+#   the scaffold had no ID and always landed in the default branch tree, so the next
+#   `sync push` created a DUPLICATE (34-config incident). With --configuration, the local file
+#   mirrors the pushed encrypted body -- placeholders would overwrite the remote on next push.
 
 # sync: GitOps -- configs as local files. init/pull/push/diff are filesystem-local (no serve REST surface).
 kbagent sync init --project ALIAS [--directory DIR] [--git-branching] [--adopt-existing]
diff --git a/plugins/kbagent/agents/keboola-expert.md b/plugins/kbagent/agents/keboola-expert.md
index 71acdc45..84fd11bf 100644
--- a/plugins/kbagent/agents/keboola-expert.md
+++ b/plugins/kbagent/agents/keboola-expert.md
@@ -129,7 +129,7 @@ been retired, so its absence is NOT a promise (see §1 Rule 6).
 | Inspect dev branch | `kbagent branch list --project P`, `kbagent branch use --project P --branch ID` | -- | acting on `main` when a dev branch exists |
 | Audit project capabilities / features | `kbagent project info --project P` -- project id, name, backend, enabled features, quota limits, metrics | -- | inspecting the UI project settings manually |
 | Manage feature flags (stack / project / user) | `kbagent feature list\|project-show\|project-add\|project-remove\|user-show\|user-add\|user-remove --project P [--email E] [--feature NAME] [--dry-run]` -- Manage API, needs a SUPER-ADMIN token (interactive prompt; `--allow-env-manage-token` for CI) | `kbagent project info` for a project's *enabled* features (read-only, no super-admin) | raw `/manage/...` calls; a manage token passed as a CLI flag |
-| Create a new config (one-shot remote, no scaffold to disk) | `kbagent config new --project P --component-id C --name N --push --no-files [--configuration @body.json]` -- default body `{}` skips validation; an explicit body is schema-validated (`--no-validate` opts out); works for every component type | `kbagent config new --output-dir D` then edit + `kbagent sync push` | raw `POST /v2/storage/components/.../configs` (no schema validation, no encryption) |
+| Create a new config (one-shot remote, no scaffold to disk) | `kbagent config new --project P --component-id C --name N --push --no-files [--configuration @body.json]` -- default body `{}` skips validation; an explicit body is schema-validated (`--no-validate` opts out); works for every component type. `--output-dir` + `--push` together is safe only on vNEXT+ (scaffold records `_keboola.config_id`, lands in the created branch's subtree); older kbagent writes an ID-less scaffold that the next `sync push` DUPLICATES (issue #644) -- there, scaffold and push in two steps | `kbagent config new --output-dir D` then edit + `kbagent sync push` | raw `POST /v2/storage/components/.../configs` (no schema validation, no encryption) |
 | Create / update / delete a config row | `kbagent config row-create\|row-update\|row-delete --project P --component-id C --config-id K [--row-id R] [--yes]` -- `row-delete` is destructive; all three are branch-aware | -- | raw REST against `/configs/K/rows` |
 | Read or write a config's runtime state | `kbagent config state-get` / `config state-set --state JSON` (0.84.2+) -- the dedicated state endpoint | -- | `config update --set 'state...'` (hard error since 0.84.2; before that it silently wrote `configuration.state.*` and left runtime state untouched) |
 | Get OAuth authorization URL | `kbagent config oauth-url --project P --component-id C --config-id K` | -- | raw `GET .../oauth/authorize` |
diff --git a/plugins/kbagent/skills/kbagent/references/commands-reference.md b/plugins/kbagent/skills/kbagent/references/commands-reference.md
index 9667eea2..33503a46 100644
--- a/plugins/kbagent/skills/kbagent/references/commands-reference.md
+++ b/plugins/kbagent/skills/kbagent/references/commands-reference.md
@@ -131,7 +131,7 @@ Requires a **super-admin** Manage API token (same kind as `org setup`). Same def
 - `config delete --project NAME --component-id ID --config-id ID [--branch ID] [--dry-run]` -- SOFT-delete a configuration into the Storage trash (restorable). Since 0.89.0 it locates the config first and a config already in the trash is NOT deleted again -- the raw API purges permanently on a second DELETE (the retry-after-timeout trap); kbagent reports `already_in_trash` and exits 0 instead. `--dry-run` reports the located state without writing. Undo with `config restore`. Permission class `destructive`
 - `config restore --project NAME --component-id ID --config-id ID [--branch ID]` *(since v0.89.0)* -- restore a trashed configuration (versions, rows and metadata come back). Only works on a config currently in the trash. Permission class `write`
 - `config trash-list --project NAME [--component-id ID] [--branch ID]` *(since v0.89.0)* -- list configurations in the trash; each row carries `component_id`, `config_id`, `name`, `version` and `deleted_at`, which is exactly what `config restore` needs. Permission class `read`
-- `config new --component-id ID [--project NAME] [--name NAME] [--output-dir DIR] [--push --no-files --description D --configuration JSON|@file|- --configuration-file PATH --no-validate --branch ID --dry-run --allow-plaintext-on-encrypt-failure]` -- **two modes**. **Default (no `--push`)**: scaffold new config from component schema; writes files to `--output-dir` or prints to stdout. **Zero API calls.** **With `--push`** (0.33.0+, requires `--project` + non-empty `--name`): also POSTs to `/v2/storage/components/{cid}/configs` for a one-shot remote create. `#`-prefixed secrets in the pushed body auto-encrypt via the Encryption API first (fail-closed; since 0.54.0, #378; `--allow-plaintext-on-encrypt-failure` overrides). `--no-files` skips the filesystem step entirely (FIIA-style empty-shell pattern). `--configuration` / `--configuration-file` override the POSTed body (default is `{}`, with validation auto-skipped for the default empty shell). `--dry-run` previews the planned POST + validation result without creating. Schema validation runs by default when an explicit body is given (fail-closed: `ConfigError` exit 5 on mismatch) but skips silently if the AI Service has no schema for the component or returns an error; `--no-validate` opts out. Works for ALL component types including `keboola.snowflake-transformation`.
+- `config new --component-id ID [--project NAME] [--name NAME] [--output-dir DIR] [--push --no-files --description D --configuration JSON|@file|- --configuration-file PATH --no-validate --branch ID --dry-run --allow-plaintext-on-encrypt-failure]` -- **two modes**. **Default (no `--push`)**: scaffold new config from component schema; writes files to `--output-dir` or prints to stdout. **Zero API calls.** **With `--push`** (0.33.0+, requires `--project` + non-empty `--name`): also POSTs to `/v2/storage/components/{cid}/configs` for a one-shot remote create. `#`-prefixed secrets in the pushed body auto-encrypt via the Encryption API first (fail-closed; since 0.54.0, #378; `--allow-plaintext-on-encrypt-failure` overrides). `--no-files` skips the filesystem step entirely (FIIA-style empty-shell pattern). `--configuration` / `--configuration-file` override the POSTed body (default is `{}`, with validation auto-skipped for the default empty shell). `--dry-run` previews the planned POST + validation result without creating. Since vNEXT, `--push` + `--output-dir` writes the scaffold WITH `_keboola.config_id` into the subtree of the branch the config was created in (registering the branch in the manifest when missing), so the next `sync push` adopts it instead of duplicating (issue #644); with an explicit body the local file mirrors the pushed encrypted configuration. Schema validation runs by default when an explicit body is given (fail-closed: `ConfigError` exit 5 on mismatch) but skips silently if the AI Service has no schema for the component or returns an error; `--no-validate` opts out. Works for ALL component types including `keboola.snowflake-transformation`.
 - `config clone --project P --component-id ID --config-id ID --name NAME [--target-project P2] [--description D] [--set PATH=VALUE ...] [--secret PATH=VALUE ...] [--branch ID] [--target-branch ID] [--dry-run] [--allow-plaintext-on-encrypt-failure]` (0.84.2+, #587) -- duplicate a configuration **whole**. Reach for this instead of reading `config detail` and rebuilding a body: copying only `configuration["parameters"]` silently drops its siblings (`runtime`, `storage`, `authorization`), and a lost `runtime.parallelism` makes Keboola fall back to `parallelism: 1` -- the reporter's 65-row writer went sequential, 140 min instead of ~60-90, with nothing in any output pointing at it. **Same project** (default): server-side copy via `POST .../configs/{id}/versions/{v}/create`; rows and `KBC::` encrypted values travel with it (verified live). `--set PATH=VALUE` is applied as a follow-up update on the copy, so an override can never be the reason a key went missing. **Cross project** (`--target-project`): reassembled client-side and rows recreated one by one, because encrypted values **cannot** travel -- a Keboola ciphertext is scoped to the project it was encrypted in. Any `KBC::` value makes the clone **fail with exit 5**, listing every path, until re-supplied via `--secret PATH=VALUE` (encrypted in the TARGET project on write). `--dry-run` reports those paths instead of refusing -- run it first to learn what to gather. Storage bucket/table IDs are copied **verbatim, never remapped**; `sync clone` is the command that remaps.
 - `config variables-set --project NAME --component-id ID --config-id ID --var KEY=VALUE [--var ...] [--replace] [--variables-id ID] [--values-id ID] [--branch ID] [--dry-run] [--allow-plaintext-on-encrypt-failure] [--yes]` -- attach variable values to a config. Auto-creates a sibling `keboola.variables` config + default row on first use and links it via the parent's `runtime.variables_id` / `variables_values_id`. Defaults to merge; `--replace` drops keys not in `--var`. `#`-prefixed values encrypt via the Encryption API (fail-closed; exit non-zero on `ENCRYPTION_FAILED`). See `variables-workflow.md`
 - `config variables-get --project NAME --component-id ID --config-id ID [--branch ID]` -- resolve `variables_id` + `values_id` from the parent config and fetch the current KEY=VALUE map. Returns `{linked: bool, variables_id, values_id, values}`; `linked=false` means the parent has no variables attached
diff --git a/plugins/kbagent/skills/kbagent/references/gotchas.md b/plugins/kbagent/skills/kbagent/references/gotchas.md
index 42d5c451..9fa18e0e 100644
--- a/plugins/kbagent/skills/kbagent/references/gotchas.md
+++ b/plugins/kbagent/skills/kbagent/references/gotchas.md
@@ -1152,6 +1152,35 @@ events and emits a final `done` SSE frame mirroring the same record.
   /openapi.json` returns the full schema, which lets the AI pick the
   right route + body shape without hard-coded knowledge.
 
+## `config new --push --output-dir` scaffolds carry the created config's ID and land in the right branch subtree (since vNEXT)
+
+- **Before vNEXT this combo was a duplicate factory** (issue #644): the
+  scaffold was written WITHOUT `_keboola.config_id` (the "assigned on first
+  push" comment was wrong on this path -- the config already existed), and it
+  always landed in the DEFAULT branch's tree even when the config was created
+  in a dev branch (`--branch`, or the active branch set by `branch create` /
+  `branch use`). The next `sync push` then classified the directory as a new
+  config and POSTed a duplicate -- a real incident created 34 duplicates.
+- **Since vNEXT** the written `_config.yml` records `_keboola.config_id`
+  (double-quoted, so legacy numeric IDs stay YAML strings) and the files are
+  written into the subtree of the branch the config was ACTUALLY created in;
+  an unregistered branch is added to `manifest.branches` exactly like
+  `sync pull --branch` would. If that registration fails, files fall back to
+  `branch-{id}/` -- never to the default tree -- with a warning in
+  `warnings[]`.
+- The follow-up `sync diff` reports the new config as `modified` (placeholder
+  scaffold vs the pushed empty body) -- that is the expected "now edit and
+  push" state, NOT drift. With an explicit `--configuration` body the local
+  directory mirrors the pushed (already encrypted) body instead, materialized
+  exactly like `sync pull` would (a transformation body's `parameters.blocks`
+  become a REAL `transform.sql`/`transform.py`, a non-empty description
+  becomes `_description.md`) -- so the diff is clean and a later push cannot
+  regress the remote to TODO placeholders.
+- `--json` gains an additive `local_scaffold: {directory, files, config_id}`
+  key on this path.
+- Agents on kbagent < vNEXT: do NOT recommend `--output-dir` + `--push`
+  together; use scaffold-only then `sync push`, or `--push --no-files`.
+
 ## `kbagent config new --push` is one-shot remote create; default is scaffold-only (since v0.33.0)
 
 - **Pre-v0.33.0**, `kbagent config new` was scaffold-only -- it wrote
diff --git a/plugins/kbagent/skills/kbagent/references/scaffold-workflow.md b/plugins/kbagent/skills/kbagent/references/scaffold-workflow.md
index daf2d213..6f263737 100644
--- a/plugins/kbagent/skills/kbagent/references/scaffold-workflow.md
+++ b/plugins/kbagent/skills/kbagent/references/scaffold-workflow.md
@@ -52,6 +52,13 @@ kbagent --json config new --component-id COMPONENT_ID --project ALIAS --name "Co
   --push --no-files
 
 # Scaffold AND remote create in one step (writes files AND POSTs)
+# Since vNEXT the written scaffold records the created config's ID
+# (_keboola.config_id) and lands in the subtree of the branch the config was
+# created in -- the next `sync push` ADOPTS the config (reported as
+# `modified` until you edit + push) instead of creating a duplicate.
+# On older versions this combo wrote an ID-less scaffold: the next
+# `sync push` DUPLICATED the config (issue #644). There, use the two-step
+# path (scaffold without --push, edit, `sync push`) instead.
 kbagent config new --component-id COMPONENT_ID --project ALIAS --name "Config Name" \
   --output-dir . --push
 ```
@@ -144,7 +151,9 @@ Rules:
 
 - `_config.yml` format follows the kbc CLI dev-friendly YAML structure
 - The `_keboola.component_id` field in `_config.yml` is required for push to work
-- `config_id` is assigned by Keboola on first push -- don't set it manually
+- `config_id` is assigned by Keboola on first push -- don't set it manually.
+  Exception: `config new --push --output-dir` (vNEXT+) writes it itself,
+  because on that path the config already exists remotely
 - Secret fields use Keboola convention: any key starting with `#` is a secret
 - Scaffold marks secret placeholders with `# encrypted by Keboola on push` comments
 - Encrypted values look like `KBC::ProjectSecure::...`
diff --git a/src/keboola_agent_cli/commands/config.py b/src/keboola_agent_cli/commands/config.py
index 17fb0212..b0b4cbbd 100644
--- a/src/keboola_agent_cli/commands/config.py
+++ b/src/keboola_agent_cli/commands/config.py
@@ -19,7 +19,9 @@
 from ..constants import KEBOOLA_DIR_NAME, MANIFEST_FILENAME, VALID_COMPONENT_TYPES
 from ..errors import ConfigError, ErrorCode, KeboolaApiError
 from ..output import format_config_detail, format_configs_table, format_search_results
+from ..services.component_service import materialize_pushed_config, stamp_scaffold_config_id
 from ..services.config_service import validate_set_paths
+from ..sync.branch_registry import default_branch_prefix
 from ._helpers import (
     check_cli_permission,
     emit_project_warnings,
@@ -33,31 +35,13 @@
 
 
 def _detect_branch_prefix(output_dir: Path) -> str | None:
-    """Detect kbc project branch path from .keboola/manifest.json.
+    """Detect the default-branch path prefix for a sync workspace.
 
-    When output_dir is inside a kbc project (has .keboola/manifest.json),
-    returns the default branch path (e.g. "main") so scaffold files
-    land in the correct location (main/extractor/... instead of extractor/...).
-
-    Returns None if not a kbc project or manifest is unreadable.
+    Delegates to :func:`..sync.branch_registry.default_branch_prefix` --
+    single source of the "scaffold under main/ when inside a kbc project"
+    convention. Returns None if not a kbc project or manifest is unreadable.
     """
-    manifest_path = output_dir / KEBOOLA_DIR_NAME / MANIFEST_FILENAME
-    if not manifest_path.is_file():
-        return None
-
-    try:
-        manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
-        branches = manifest.get("branches", [])
-        if branches:
-            # Use the first (default) branch path
-            branch_path = branches[0].get("path", "")
-            if branch_path:
-                logger.debug("Detected kbc branch prefix: %s", branch_path)
-                return branch_path
-    except (json.JSONDecodeError, OSError) as exc:
-        logger.debug("Could not read manifest: %s", exc)
-
-    return None
+    return default_branch_prefix(output_dir)
 
 
 config_app = typer.Typer(help="Browse and inspect configurations")
@@ -1419,6 +1403,13 @@ def config_new(
         # POST (the "scaffold + push" combo). Dry-run must NOT touch the
         # filesystem -- the user expects a preview, not a side effect.
         #
+        # Issue #644: the config exists remotely at this point, so the file
+        # must carry its identity (``_keboola.config_id``) and must land in
+        # the subtree of the branch the config was actually created in --
+        # otherwise the next ``sync push`` creates a duplicate (dev-branch
+        # creates used to scaffold into ``main/``, invisible to the dev push
+        # and a phantom "added" for the production push).
+        #
         # ``silent=True``: the push-result envelope below is the authoritative
         # output for this path. In JSON mode emitting the scaffold envelope
         # too would produce two concatenated JSON documents on stdout (breaks
@@ -1427,7 +1418,50 @@ def config_new(
         # passed as a sentinel ``False`` because ``silent=True`` short-circuits
         # before it is read.
         if output_dir and scaffold is not None and not dry_run:
-            _write_scaffold_to_disk(formatter, scaffold, output_dir, json_mode=False, silent=True)
+            created_id = str(push_result.get("id") or "")
+            placement = get_service(ctx, "sync_service").resolve_scaffold_placement(
+                alias=project or "",
+                project_root=Path(output_dir),
+                branch_id=push_result.get("branch_id"),
+            )
+            pushed_body = push_result.get("configuration")
+            if created_id and config_body is not None and isinstance(pushed_body, dict):
+                # An explicit body was pushed: mirror it exactly the way
+                # ``sync pull`` would materialize it (real code extracted to
+                # transform.sql/.py, not placeholder templates) -- writing the
+                # placeholder scaffold here would make the next ``sync push``
+                # overwrite the real remote config with TODOs.
+                target_dir = _scaffold_target_dir(
+                    output_dir, placement.branch_prefix, scaffold["directory"]
+                )
+                files_written = materialize_pushed_config(
+                    component_id=component_id,
+                    config_id=created_id,
+                    name=str(push_result.get("name") or name),
+                    description=str(push_result.get("description") or ""),
+                    configuration=pushed_body,
+                    config_dir=target_dir,
+                )
+                written_dir = target_dir
+            else:
+                if created_id:
+                    scaffold = stamp_scaffold_config_id(scaffold, created_id)
+                written_dir = _write_scaffold_to_disk(
+                    formatter,
+                    scaffold,
+                    output_dir,
+                    json_mode=False,
+                    silent=True,
+                    branch_prefix=placement.branch_prefix,
+                )
+                files_written = [f["path"] for f in scaffold["files"]]
+            push_result["local_scaffold"] = {
+                "directory": str(written_dir),
+                "files": files_written,
+                "config_id": created_id or None,
+            }
+            if placement.warning:
+                push_result.setdefault("warnings", []).append(placement.warning)
 
         if formatter.json_mode:
             formatter.output(push_result)
@@ -1438,7 +1472,13 @@ def config_new(
     # ── Scaffold-only path: today's behavior, byte-for-byte unchanged ────────
     assert scaffold is not None  # narrowing for type-checker; skip_scaffold is False here.
     if output_dir:
-        _write_scaffold_to_disk(formatter, scaffold, output_dir, json_mode=formatter.json_mode)
+        _write_scaffold_to_disk(
+            formatter,
+            scaffold,
+            output_dir,
+            json_mode=formatter.json_mode,
+            branch_prefix=_detect_branch_prefix(Path(output_dir)),
+        )
     else:
         # Print scaffold content
         if formatter.json_mode:
@@ -1466,17 +1506,29 @@ def config_new(
                 formatter.console.print()
 
 
+def _scaffold_target_dir(output_dir: str, branch_prefix: str | None, scaffold_dir: str) -> Path:
+    """Join the output dir, optional branch subtree, and scaffold directory."""
+    if branch_prefix:
+        return Path(output_dir) / branch_prefix / scaffold_dir
+    return Path(output_dir) / scaffold_dir
+
+
 def _write_scaffold_to_disk(
     formatter: Any,
     scaffold: dict[str, Any],
     output_dir: str,
     json_mode: bool,
     silent: bool = False,
-) -> None:
+    branch_prefix: str | None = None,
+) -> Path:
     """Shared helper: write the generated scaffold files under ``output_dir``.
 
-    Detects an enclosing ``main/`` branch prefix the same way the pre-push
-    path does, so the layout matches what ``kbagent sync push`` would expect.
+    ``branch_prefix`` selects the branch subtree (``None`` = flat layout).
+    The scaffold-only path passes :func:`_detect_branch_prefix`'s result
+    (historical behaviour -- default branch path or flat); the ``--push``
+    path passes ``ScaffoldPlacement.branch_prefix``, because a dev-branch
+    create must land in the dev branch's subtree (issue #644). Returns the
+    directory the files were written into.
 
     Output rules:
     - ``silent=True``: write files only; emit no banner and no JSON envelope.
@@ -1488,22 +1540,19 @@ def _write_scaffold_to_disk(
       ``{directory, files_written}`` when ``json_mode`` is set, otherwise a
       dim-formatted "Scaffold written to ..." banner.
     """
-    branch_prefix = _detect_branch_prefix(Path(output_dir))
-    if branch_prefix:
-        scaffold_dir = branch_prefix + "/" + scaffold["directory"]
-    else:
-        scaffold_dir = scaffold["directory"]
-
-    base_path = Path(output_dir) / scaffold_dir
+    base_path = _scaffold_target_dir(output_dir, branch_prefix, scaffold["directory"])
     base_path.mkdir(parents=True, exist_ok=True)
 
     for file_entry in scaffold["files"]:
         file_path = base_path / file_entry["path"]
         file_path.parent.mkdir(parents=True, exist_ok=True)
-        file_path.write_text(file_entry["content"], encoding="utf-8")
+        # newline="" so Windows does not CRLF-translate scaffold files that
+        # sit next to LF-only files written by sync pull (same convention as
+        # SyncService._write_config_file).
+        file_path.write_text(file_entry["content"], encoding="utf-8", newline="")
 
     if silent:
-        return
+        return base_path
 
     if json_mode:
         formatter.output(
@@ -1516,6 +1565,7 @@ def _write_scaffold_to_disk(
         formatter.console.print(
             f"[dim]Scaffold written to {base_path} ({len(scaffold['files'])} file(s))[/dim]"
         )
+    return base_path
 
 
 def _render_push_result_human(
@@ -1552,6 +1602,15 @@ def _render_push_result_human(
     formatter.success(
         f"Created config '{escape(name)}' [{config_id}] in {escape(component_id)}{branch_info}"
     )
+    local_scaffold = result.get("local_scaffold")
+    if local_scaffold:
+        id_note = ", config_id recorded" if local_scaffold.get("config_id") else ""
+        formatter.console.print(
+            f"[dim]Files written to {escape(local_scaffold['directory'])} "
+            f"({len(local_scaffold['files'])} file(s){id_note})[/dim]"
+        )
+    for warning in result.get("warnings", []) or []:
+        formatter.console.print(f"[yellow]⚠ {escape(warning)}[/yellow]")
     if validation_status == "skipped":
         formatter.console.print(
             "[dim]Note: schema validation was skipped "
diff --git a/src/keboola_agent_cli/commands/context.py b/src/keboola_agent_cli/commands/context.py
index 42193cbd..8ed69437 100644
--- a/src/keboola_agent_cli/commands/context.py
+++ b/src/keboola_agent_cli/commands/context.py
@@ -427,7 +427,14 @@
     entirely for FIIA-style one-shot creates. Schema validation runs by default when an explicit
     --configuration body is given (fail-closed; --no-validate opts out). Default body is {{}}
     (empty shell, validation auto-skipped). Works for ALL component types including
-    keboola.snowflake-transformation.
+    keboola.snowflake-transformation. With --push AND --output-dir the written scaffold
+    records the created config's identity (_keboola.config_id) and lands in the subtree
+    of the branch the config was created in (--branch or the active branch; the branch
+    is registered in the manifest if missing) -- the next sync push adopts the config
+    instead of creating a duplicate. With an explicit --configuration body the local
+    directory mirrors the pushed (already encrypted) body exactly like sync pull would
+    materialize it (real transform.sql/.py extracted from the body) instead of
+    placeholder scaffolding.
 
   kbagent config clone --project P --component-id ID --config-id ID --name NAME
                        [--target-project P2] [--description D] [--set PATH=VALUE ...]
diff --git a/src/keboola_agent_cli/services/component_service.py b/src/keboola_agent_cli/services/component_service.py
index 6e3d9d09..7f7c65b8 100644
--- a/src/keboola_agent_cli/services/component_service.py
+++ b/src/keboola_agent_cli/services/component_service.py
@@ -7,15 +7,18 @@
 
 import logging
 from collections.abc import Callable
+from pathlib import Path
 from typing import Any
 
 import yaml
 
 from ..ai_client import AiServiceClient
 from ..config_store import ConfigStore
-from ..constants import SECRET_PLACEHOLDER
+from ..constants import CONFIG_FILENAME, SECRET_PLACEHOLDER
 from ..errors import ConfigError, KeboolaApiError
 from ..models import ComponentDetail, ComponentSuggestion, ProjectConfig
+from ..sync.code_extraction import DESCRIPTION_FILENAME, extract_code_files
+from ..sync.config_format import api_config_to_local, dump_config_yaml
 from .base import BaseService, ClientFactory
 from .org_service import slugify
 
@@ -293,6 +296,121 @@ def _build_pyproject_toml(component_id: str, name: str, packages: list[str] | No
     )
 
 
+_SCAFFOLD_ID_NOTE = "# NOTE: config_id will be assigned by Keboola on first push"
+_SCAFFOLD_ID_STAMPED_NOTE = (
+    "# NOTE: created remotely by 'config new --push'; config_id recorded below"
+)
+
+
+def stamp_scaffold_config_id(scaffold: dict[str, Any], config_id: str) -> dict[str, Any]:
+    """Return a copy of *scaffold* whose ``_config.yml`` records *config_id*.
+
+    On the ``config new --push --output-dir`` path the configuration already
+    exists remotely by the time the scaffold hits the disk. Writing the file
+    without ``_keboola.config_id`` makes the next ``sync push`` classify the
+    directory as a brand-new configuration and create a duplicate
+    (issue #644). With the ID present, the sync diff's adopt-by-id guard
+    (issue #482) pairs the directory with the existing remote config instead.
+
+    The ID is emitted double-quoted so legacy numeric IDs stay YAML strings
+    (an unquoted ``12345`` would parse as ``int`` and never match the
+    string-keyed remote lookup). A ``_config.yml`` without a ``_keboola``
+    block (flow scaffolds, issue #650) gets one appended wholesale. Pure
+    function: the input scaffold is not mutated.
+    """
+    files: list[dict[str, Any]] = []
+    for entry in scaffold["files"]:
+        if entry["path"] != CONFIG_FILENAME:
+            # Copy, don't share -- a later in-place mutation of a companion
+            # entry must not reach back into the caller's scaffold.
+            files.append(dict(entry))
+            continue
+        content: str = entry["content"]
+        out: list[str] = []
+        in_keboola = False
+        stamped = False
+        for line in content.splitlines():
+            if line == _SCAFFOLD_ID_NOTE:
+                out.append(_SCAFFOLD_ID_STAMPED_NOTE)
+                continue
+            out.append(line)
+            if line == "_keboola:":
+                in_keboola = True
+            elif in_keboola and line.startswith("  component_id:"):
+                out.append(f'  config_id: "{config_id}"')
+                stamped = True
+                in_keboola = False
+        if not stamped:
+            if out and out[-1] != "":
+                out.append("")
+            out.extend(
+                [
+                    "_keboola:",
+                    f"  component_id: {scaffold['component_id']}",
+                    f'  config_id: "{config_id}"',
+                ]
+            )
+        new_content = "\n".join(out)
+        if content.endswith("\n"):
+            new_content += "\n"
+        files.append({**entry, "content": new_content})
+    return {**scaffold, "files": files}
+
+
+def materialize_pushed_config(
+    component_id: str,
+    config_id: str,
+    name: str,
+    description: str,
+    configuration: dict[str, Any],
+    config_dir: Path,
+) -> list[str]:
+    """Write the local files mirroring an explicitly pushed config body.
+
+    Used by ``config new --push --output-dir`` when ``--configuration`` /
+    ``--configuration-file`` supplied a real body: writing the placeholder
+    scaffold instead would diverge from the remote, and the next ``sync
+    push`` would overwrite the freshly created configuration with TODO
+    templates. The mirrored body comes from the API response, so
+    ``#``-secrets are already encrypted (``KBC::...``) -- no plaintext ever
+    reaches the disk.
+
+    The directory is materialized exactly the way ``sync pull`` would do it:
+    the same converter (:func:`api_config_to_local`), the same code
+    extraction (a transformation body's ``parameters.blocks`` become a real
+    ``transform.sql`` / ``transform.py`` next to ``_config.yml`` -- NOT
+    placeholder templates, which would be merged back over the real code on
+    the next push), and the same YAML serialization
+    (:func:`..sync.config_format.dump_config_yaml`, written with
+    ``newline=""`` so Windows does not CRLF-translate what every other sync
+    write leaves LF-only). The follow-up ``sync diff`` therefore reports no
+    spurious changes.
+
+    Returns the list of file paths written, relative to *config_dir*.
+    """
+    local = api_config_to_local(
+        component_id,
+        {"name": name, "description": description, "configuration": configuration},
+        str(config_id),
+    )
+    config_dir.mkdir(parents=True, exist_ok=True)
+    # The slugified directory can pre-exist (same-name re-run, stray files):
+    # report only what THIS call wrote, and clear a stale _description.md a
+    # previous occupant left behind when the pushed description is empty --
+    # _extract_description only ever writes, so the stale file would
+    # misattribute to the new config_id (PR #653 review sweep).
+    if not description:
+        stale_description = config_dir / DESCRIPTION_FILENAME
+        stale_description.unlink(missing_ok=True)
+    before = {q for q in config_dir.rglob("*") if q.is_file()}
+    extract_code_files(component_id, local, config_dir)
+    content = dump_config_yaml(local)
+    (config_dir / CONFIG_FILENAME).write_text(content, encoding="utf-8", newline="")
+    after = {q for q in config_dir.rglob("*") if q.is_file()}
+    created = (after - before) | {config_dir / CONFIG_FILENAME}
+    return sorted(q.relative_to(config_dir).as_posix() for q in created)
+
+
 def _build_flow_config_yml(name: str, component_id: str = "keboola.flow") -> str:
     """Generate a conditional-flow (keboola.flow) configuration YAML skeleton.
 
diff --git a/src/keboola_agent_cli/services/sync_service.py b/src/keboola_agent_cli/services/sync_service.py
index e1f3cd36..ec665da1 100644
--- a/src/keboola_agent_cli/services/sync_service.py
+++ b/src/keboola_agent_cli/services/sync_service.py
@@ -25,11 +25,18 @@
     MANIFEST_VERSION,
 )
 from ..errors import ConfigError, ErrorCode, KeboolaApiError, SyncConflictError
+from ..sync.branch_registry import (
+    ScaffoldPlacement,
+    ensure_branch_registered,
+    register_branch_dir,
+    resolve_scaffold_placement,
+)
 from ..sync.code_extraction import extract_code_files, merge_code_files
 from ..sync.config_format import (
     api_config_to_local,
     api_row_to_local,
     classify_component_type,
+    dump_config_yaml,
     local_config_to_api,
     local_row_to_api,
 )
@@ -2089,13 +2096,7 @@ def _write_config_file(self, config_dir: Path, config_data: dict[str, Any]) -> s
         """
         config_dir.mkdir(parents=True, exist_ok=True)
         config_file = config_dir / CONFIG_FILENAME
-        content = yaml.dump(
-            config_data,
-            default_flow_style=False,
-            allow_unicode=True,
-            sort_keys=False,
-            width=120,
-        )
+        content = dump_config_yaml(config_data)
         config_file.write_text(content, encoding="utf-8", newline="")
         return hashlib.sha256(content.encode("utf-8")).hexdigest()
 
@@ -2174,50 +2175,44 @@ def _read_config_file(self, config_dir: Path) -> dict[str, Any] | None:
             logger.warning("Failed to parse %s", config_file)
             return None
 
+    def register_branch_dir(
+        self,
+        alias: str,
+        project_root: Path,
+        branch_id: int,
+    ) -> str:
+        """Resolve (and register if needed) the on-disk directory for *branch_id*.
+
+        Thin wrapper over :func:`..sync.branch_registry.register_branch_dir`
+        (issue #644); see that module for the semantics.
+        """
+        projects = self.resolve_projects([alias])
+        return register_branch_dir(projects[alias], project_root, branch_id, self._client_factory)
+
+    def resolve_scaffold_placement(
+        self,
+        alias: str,
+        project_root: Path,
+        branch_id: int | None,
+    ) -> ScaffoldPlacement:
+        """Resolve where a ``config new --push`` scaffold belongs on disk.
+
+        Thin wrapper over :func:`..sync.branch_registry.resolve_scaffold_placement`
+        (issue #644).
+        """
+        # Resolved for BOTH paths: the production path needs the project id
+        # for the foreign-workspace mismatch check (PR #653 review sweep).
+        project = self.resolve_projects([alias])[alias]
+        return resolve_scaffold_placement(project, project_root, branch_id, self._client_factory)
+
     def _ensure_branch_registered(
         self,
         manifest: Manifest,
         branch_id: int | None,
         client: Any,
     ) -> str | None:
-        """Ensure *branch_id* has an entry in ``manifest.branches``.
-
-        If *branch_id* is ``None`` (production) or already present, this is
-        a no-op.  Otherwise the branch name is fetched from the API and a
-        new :class:`ManifestBranch` is appended.
-
-        Returns:
-            The new branch path if one was added, ``None`` otherwise.
-        """
-        if branch_id is None:
-            return None
-
-        # Already registered?
-        for branch in manifest.branches:
-            if branch.id == branch_id:
-                return None
-
-        # Fetch branch info from API to get a human-readable name
-        all_branches = client.list_dev_branches()
-        branch_name = ""
-        for b in all_branches:
-            if b.get("id") == branch_id:
-                branch_name = b.get("name", "")
-                break
-
-        # Generate filesystem-safe path
-        path = sanitize_name(branch_name) if branch_name else ""
-        if not path:
-            path = f"branch-{branch_id}"
-
-        # Handle path uniqueness -- avoid collisions with existing entries
-        existing_paths = {br.path for br in manifest.branches}
-        if path in existing_paths:
-            path = f"{path}-{branch_id}"
-
-        manifest.branches.append(ManifestBranch(id=branch_id, path=path))
-        logger.info("Registered dev branch %d as '%s' in manifest", branch_id, path)
-        return path
+        """Delegate to :func:`..sync.branch_registry.ensure_branch_registered`."""
+        return ensure_branch_registered(manifest, branch_id, client)
 
     def _find_branch_path(self, manifest: Manifest, branch_id: int | None) -> str:
         """Find the branch directory name for a given branch ID.
diff --git a/src/keboola_agent_cli/sync/branch_registry.py b/src/keboola_agent_cli/sync/branch_registry.py
new file mode 100644
index 00000000..b59995c2
--- /dev/null
+++ b/src/keboola_agent_cli/sync/branch_registry.py
@@ -0,0 +1,244 @@
+"""Branch directory registration for sync workspaces.
+
+Extracted from ``SyncService`` (file-size budget): resolving/registering the
+``manifest.branches`` entry for a dev branch is needed both by ``sync pull``
+/ ``sync diff`` (ensure the pulled branch has a directory) and by
+``config new --push --output-dir`` (issue #644: a config created in a dev
+branch must scaffold into that branch's subtree, never into the default
+branch's tree).
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+from ..constants import KEBOOLA_DIR_NAME, MANIFEST_FILENAME
+from ..errors import ConfigError
+from .manifest import Manifest, ManifestBranch, load_manifest, save_manifest
+from .naming import sanitize_name
+
+logger = logging.getLogger(__name__)
+
+
+def fallback_branch_dir(branch_id: int) -> str:
+    """Canonical directory name for a branch whose real name is unknown.
+
+    Single source of the ``branch-{id}`` convention -- one spelling
+    everywhere means a later ``sync pull --branch`` lands in the same
+    directory a fallback scaffold was written to.
+    """
+    return f"branch-{branch_id}"
+
+
+def default_branch_prefix(project_root: Path) -> str | None:
+    """Tolerant read of the default branch's directory from the manifest.
+
+    Returns ``None`` when *project_root* is not a sync workspace or the
+    manifest is unreadable -- callers then use a flat layout. Deliberately a
+    raw JSON peek, not :func:`load_manifest`: the historical
+    ``_detect_branch_prefix`` behaviour of ``config new`` tolerates partial
+    manifests (e.g. hand-written or older shapes missing optional sections),
+    and a read-only prefix lookup has no business schema-validating the
+    whole file.
+    """
+    raw = _peek_manifest(project_root)
+    if raw:
+        branches = raw.get("branches", [])
+        if branches:
+            return branches[0].get("path") or None
+    return None
+
+
+def _peek_manifest(project_root: Path) -> dict[str, Any] | None:
+    """Raw, tolerant read of ``.keboola/manifest.json`` (None on any problem)."""
+    manifest_path = project_root / KEBOOLA_DIR_NAME / MANIFEST_FILENAME
+    if not manifest_path.is_file():
+        return None
+    try:
+        loaded = json.loads(manifest_path.read_text(encoding="utf-8"))
+        return loaded if isinstance(loaded, dict) else None
+    except (json.JSONDecodeError, OSError) as exc:
+        logger.debug("Could not read manifest under %s: %s", project_root, exc)
+        return None
+
+
+@dataclass(frozen=True)
+class ScaffoldPlacement:
+    """Where a pushed scaffold's files belong, plus an optional warning.
+
+    ``branch_prefix`` is the branch directory relative to the output dir
+    (``None`` = flat layout, no sync workspace). ``warning`` is set when
+    placement degraded (branch registration failed) and should be surfaced
+    to the user.
+    """
+
+    branch_prefix: str | None
+    warning: str | None = None
+
+
+def resolve_scaffold_placement(
+    project: Any,
+    project_root: Path,
+    branch_id: int | None,
+    client_factory: Any,
+) -> ScaffoldPlacement:
+    """Resolve the branch subtree a pushed scaffold must be written into.
+
+    - Production create (``branch_id is None``): the default branch's
+      directory when *project_root* is a sync workspace, flat otherwise
+      (pre-#644 behaviour).
+    - Dev-branch create outside a sync workspace: flat layout.
+    - Dev-branch create inside a sync workspace: the branch's directory
+      from the manifest, registering the branch first when unknown
+      (:func:`register_branch_dir`). On any failure the files fall back to
+      ``branch-{id}/`` with a warning -- NEVER to the default branch tree,
+      because a wrong-branch file is exactly the duplicate factory issue
+      #644 describes. This includes a workspace belonging to a different
+      project: the config was already created remotely, so the files are
+      still written (inert, untracked by that workspace's manifest) rather
+      than lost, and the warning names the mismatch.
+    """
+    if branch_id is None:
+        mismatch = _project_mismatch_warning(project, project_root)
+        if mismatch:
+            # Foreign workspace: write FLAT (outside every branch tree, so a
+            # later `sync push` there can never pick the files up as a new
+            # config of the WRONG project) and say so.
+            return ScaffoldPlacement(None, mismatch)
+        return ScaffoldPlacement(default_branch_prefix(project_root))
+    if not (project_root / KEBOOLA_DIR_NAME / MANIFEST_FILENAME).is_file():
+        return ScaffoldPlacement(None)
+    try:
+        return ScaffoldPlacement(
+            register_branch_dir(project, project_root, int(branch_id), client_factory)
+        )
+    except Exception as exc:
+        fallback = fallback_branch_dir(branch_id)
+        warning = (
+            f"Could not resolve the manifest directory for branch {branch_id} "
+            f"({exc}); scaffold written under '{fallback}/'. Run "
+            f"'kbagent sync pull --branch {branch_id}' to reconcile."
+        )
+        logger.warning(warning)
+        return ScaffoldPlacement(fallback, warning)
+
+
+def ensure_branch_registered(
+    manifest: Manifest,
+    branch_id: int | None,
+    client: Any,
+) -> str | None:
+    """Ensure *branch_id* has an entry in ``manifest.branches``.
+
+    If *branch_id* is ``None`` (production) or already present, this is
+    a no-op.  Otherwise the branch name is fetched from the API and a
+    new :class:`ManifestBranch` is appended.
+
+    Returns:
+        The new branch path if one was added, ``None`` otherwise.
+    """
+    if branch_id is None:
+        return None
+
+    # Already registered?
+    for branch in manifest.branches:
+        if branch.id == branch_id:
+            return None
+
+    # Fetch branch info from API to get a human-readable name
+    all_branches = client.list_dev_branches()
+    branch_name = ""
+    for b in all_branches:
+        if b.get("id") == branch_id:
+            branch_name = b.get("name", "")
+            break
+
+    # Generate filesystem-safe path
+    path = sanitize_name(branch_name) if branch_name else ""
+    if not path:
+        path = f"branch-{branch_id}"
+
+    # Handle path uniqueness -- avoid collisions with existing entries
+    existing_paths = {br.path for br in manifest.branches}
+    if path in existing_paths:
+        path = f"{path}-{branch_id}"
+
+    manifest.branches.append(ManifestBranch(id=branch_id, path=path))
+    logger.info("Registered dev branch %d as '%s' in manifest", branch_id, path)
+    return path
+
+
+def register_branch_dir(
+    project: Any,
+    project_root: Path,
+    branch_id: int,
+    client_factory: Any,
+) -> str:
+    """Resolve (and register if needed) the on-disk directory for *branch_id*.
+
+    Used by ``config new --push --output-dir`` (issue #644): a config
+    created in a dev branch must scaffold into that branch's subtree, not
+    into the default branch's tree -- a wrong-branch file is invisible to
+    the dev-branch push and later duplicates the config on a production
+    push. When the branch is not yet in ``manifest.branches``, it is
+    registered exactly the way ``sync pull --branch`` would
+    (:func:`ensure_branch_registered`: branch name fetched from the API,
+    sanitized path, ``branch-{id}`` fallback), so a later pull reuses the
+    same directory.
+
+    Args:
+        project: Resolved project config (``stack_url``, ``token``,
+            ``project_id`` attributes).
+        project_root: Sync workspace root (must contain ``.keboola/``).
+        branch_id: Dev branch the configuration was created in.
+        client_factory: ``(stack_url, token) -> KeboolaClient`` callable.
+
+    Raises:
+        ConfigError: manifest missing/unreadable, or the manifest belongs
+            to a different project than *project*.
+        KeboolaApiError: branch-name lookup failed.
+    """
+    manifest = load_manifest(project_root)
+    if project.project_id is not None and manifest.project.id != project.project_id:
+        raise ConfigError(
+            f"Manifest in {project_root} belongs to project {manifest.project.id}, "
+            f"not to project {project.project_id}"
+        )
+    for branch in manifest.branches:
+        if branch.id == branch_id:
+            return branch.path
+    client = client_factory(project.stack_url, project.token)
+    with client:
+        path = ensure_branch_registered(manifest, branch_id, client)
+    save_manifest(project_root, manifest)
+    if path:
+        return path
+    # ensure_branch_registered returned None despite the pre-check miss --
+    # defensive re-lookup so the caller always gets a directory.
+    for branch in manifest.branches:
+        if branch.id == branch_id:
+            return branch.path
+    return f"branch-{branch_id}"
+
+
+def _project_mismatch_warning(project: Any, project_root: Path) -> str | None:
+    """Warning text when *project_root*'s manifest belongs to another project.
+
+    ``None`` when the identities match or either side is unknown (no
+    manifest, unreadable manifest, or a project without a stored id).
+    """
+    if project is None or getattr(project, "project_id", None) is None:
+        return None
+    raw = _peek_manifest(project_root)
+    manifest_project_id = (raw or {}).get("project", {}).get("id")
+    if manifest_project_id is None or manifest_project_id == project.project_id:
+        return None
+    return (
+        f"Manifest in {project_root} belongs to project {manifest_project_id}, "
+        f"not to project {project.project_id}; scaffold written flat at the "
+        f"workspace root (untracked by that workspace's sync tree)."
+    )
diff --git a/src/keboola_agent_cli/sync/code_extraction.py b/src/keboola_agent_cli/sync/code_extraction.py
index 03d6375e..33742a1a 100644
--- a/src/keboola_agent_cli/sync/code_extraction.py
+++ b/src/keboola_agent_cli/sync/code_extraction.py
@@ -313,7 +313,9 @@ def merge_code_files(
 
 def _extract_sql_transformation(config_data: dict[str, Any], config_dir: Path) -> dict[str, Any]:
     """Extract SQL blocks from parameters.blocks into transform.sql."""
-    parameters = config_data.get("parameters", {})
+    parameters = config_data.get("parameters") or {}
+    if not isinstance(parameters, dict):
+        return config_data
     blocks = parameters.get("blocks", [])
 
     if not blocks:
@@ -439,7 +441,9 @@ def _parse_sql_blocks(content: str) -> list[dict[str, Any]]:
 
 def _extract_python_transformation(config_data: dict[str, Any], config_dir: Path) -> dict[str, Any]:
     """Extract Python blocks from parameters.blocks into transform.py, packages into pyproject.toml."""
-    parameters = config_data.get("parameters", {})
+    parameters = config_data.get("parameters") or {}
+    if not isinstance(parameters, dict):
+        return config_data
     blocks = parameters.get("blocks", [])
 
     if blocks:
@@ -553,7 +557,9 @@ def _parse_python_blocks(content: str) -> list[dict[str, Any]]:
 
 def _extract_python_app(config_data: dict[str, Any], config_dir: Path) -> dict[str, Any]:
     """Extract parameters.code into code.py and packages into pyproject.toml."""
-    parameters = config_data.get("parameters", {})
+    parameters = config_data.get("parameters") or {}
+    if not isinstance(parameters, dict):
+        return config_data
 
     code = parameters.get("code")
     if code and isinstance(code, str):
diff --git a/src/keboola_agent_cli/sync/config_format.py b/src/keboola_agent_cli/sync/config_format.py
index da5e4e45..e71c36fc 100644
--- a/src/keboola_agent_cli/sync/config_format.py
+++ b/src/keboola_agent_cli/sync/config_format.py
@@ -10,6 +10,8 @@
 import copy
 from typing import Any
 
+import yaml
+
 from ..constants import CONFIG_YML_VERSION
 
 
@@ -296,3 +298,25 @@ def local_row_to_api(
             configuration.setdefault(key, value)
 
     return name, description, configuration
+
+
+# ---------------------------------------------------------------------------
+# Canonical _config.yml serialization
+# ---------------------------------------------------------------------------
+
+
+def dump_config_yaml(config_data: dict[str, Any]) -> str:
+    """Serialize a local ``_config.yml`` dict with the canonical settings.
+
+    The single source of truth for how sync materializes ``_config.yml``
+    content -- ``SyncService._write_config_file`` (pull) and
+    ``config new --push --output-dir`` (mirrored-body scaffold, issue #644)
+    both call this, so the two paths can never drift in formatting.
+    """
+    return yaml.dump(
+        config_data,
+        default_flow_style=False,
+        allow_unicode=True,
+        sort_keys=False,
+        width=120,
+    )
diff --git a/tests/test_branch_registry.py b/tests/test_branch_registry.py
new file mode 100644
index 00000000..14f870d7
--- /dev/null
+++ b/tests/test_branch_registry.py
@@ -0,0 +1,165 @@
+"""Unit tests for sync/branch_registry.py (issue #644).
+
+Covers the placement policy that decides where a ``config new --push``
+scaffold lands: default-branch prefix for production creates, branch
+subtree (with on-demand registration) for dev-branch creates, and the
+``branch-{id}/`` degrade path that must NEVER retarget files to the
+default tree.
+"""
+
+import json
+from pathlib import Path
+from unittest.mock import MagicMock
+
+import pytest
+
+from keboola_agent_cli.errors import ConfigError
+from keboola_agent_cli.sync.branch_registry import (
+    ScaffoldPlacement,
+    default_branch_prefix,
+    ensure_branch_registered,
+    fallback_branch_dir,
+    register_branch_dir,
+    resolve_scaffold_placement,
+)
+from keboola_agent_cli.sync.manifest import load_manifest
+
+
+def _write_manifest(root: Path, branches: list[dict], project_id: int = 1234) -> None:
+    keboola = root / ".keboola"
+    keboola.mkdir(parents=True, exist_ok=True)
+    manifest = {
+        "version": 2,
+        "project": {"id": project_id, "apiHost": "connection.keboola.com"},
+        "naming": {},
+        "branches": branches,
+        "configurations": [],
+    }
+    (keboola / "manifest.json").write_text(json.dumps(manifest), encoding="utf-8")
+
+
+def _project(project_id: int | None = 1234) -> MagicMock:
+    project = MagicMock()
+    project.stack_url = "https://connection.keboola.com"
+    project.token = "901-55555-fakeTestTokenDoNotUseXXXXXXXX"
+    project.project_id = project_id
+    return project
+
+
+def _client_factory_returning(branches: list[dict]) -> MagicMock:
+    client = MagicMock()
+    client.list_dev_branches.return_value = branches
+    client.__enter__ = MagicMock(return_value=client)
+    client.__exit__ = MagicMock(return_value=False)
+    factory = MagicMock(return_value=client)
+    return factory
+
+
+class TestFallbackBranchDir:
+    def test_canonical_spelling(self) -> None:
+        assert fallback_branch_dir(51406) == "branch-51406"
+
+
+class TestDefaultBranchPrefix:
+    def test_no_manifest_returns_none(self, tmp_path: Path) -> None:
+        assert default_branch_prefix(tmp_path) is None
+
+    def test_returns_first_branch_path(self, tmp_path: Path) -> None:
+        _write_manifest(tmp_path, [{"id": 10, "path": "main"}, {"id": 20, "path": "dev"}])
+        assert default_branch_prefix(tmp_path) == "main"
+
+    def test_unreadable_manifest_returns_none(self, tmp_path: Path) -> None:
+        keboola = tmp_path / ".keboola"
+        keboola.mkdir()
+        (keboola / "manifest.json").write_text("{not json", encoding="utf-8")
+        assert default_branch_prefix(tmp_path) is None
+
+
+class TestRegisterBranchDir:
+    def test_already_registered_returns_path_without_client(self, tmp_path: Path) -> None:
+        _write_manifest(tmp_path, [{"id": 10, "path": "main"}, {"id": 20, "path": "dev-x"}])
+        factory = MagicMock()
+        assert register_branch_dir(_project(), tmp_path, 20, factory) == "dev-x"
+        factory.assert_not_called()
+
+    def test_registers_unknown_branch_and_persists(self, tmp_path: Path) -> None:
+        _write_manifest(tmp_path, [{"id": 10, "path": "main"}])
+        factory = _client_factory_returning([{"id": 20, "name": "Feature X"}])
+        path = register_branch_dir(_project(), tmp_path, 20, factory)
+        assert path == "feature-x"
+        manifest = load_manifest(tmp_path)
+        assert [(b.id, b.path) for b in manifest.branches] == [(10, "main"), (20, "feature-x")]
+
+    def test_project_mismatch_raises(self, tmp_path: Path) -> None:
+        _write_manifest(tmp_path, [{"id": 10, "path": "main"}], project_id=9999)
+        with pytest.raises(ConfigError, match="belongs to project 9999"):
+            register_branch_dir(_project(1234), tmp_path, 20, MagicMock())
+
+
+class TestEnsureBranchRegistered:
+    def test_unknown_name_falls_back_to_branch_id_dir(self, tmp_path: Path) -> None:
+        _write_manifest(tmp_path, [{"id": 10, "path": "main"}])
+        manifest = load_manifest(tmp_path)
+        client = MagicMock()
+        client.list_dev_branches.return_value = []  # API knows nothing
+        assert ensure_branch_registered(manifest, 20, client) == "branch-20"
+
+
+class TestResolveScaffoldPlacement:
+    def test_production_create_uses_default_prefix(self, tmp_path: Path) -> None:
+        _write_manifest(tmp_path, [{"id": 10, "path": "main"}])
+        placement = resolve_scaffold_placement(None, tmp_path, None, MagicMock())
+        assert placement == ScaffoldPlacement("main")
+
+    def test_production_create_flat_without_manifest(self, tmp_path: Path) -> None:
+        placement = resolve_scaffold_placement(None, tmp_path, None, MagicMock())
+        assert placement == ScaffoldPlacement(None)
+
+    def test_branch_create_without_manifest_is_flat(self, tmp_path: Path) -> None:
+        placement = resolve_scaffold_placement(_project(), tmp_path, 20, MagicMock())
+        assert placement == ScaffoldPlacement(None)
+
+    def test_branch_create_registers_and_places(self, tmp_path: Path) -> None:
+        _write_manifest(tmp_path, [{"id": 10, "path": "main"}])
+        factory = _client_factory_returning([{"id": 20, "name": "Feature X"}])
+        placement = resolve_scaffold_placement(_project(), tmp_path, 20, factory)
+        assert placement == ScaffoldPlacement("feature-x")
+
+    def test_registration_failure_degrades_to_branch_dir_with_warning(self, tmp_path: Path) -> None:
+        """NEVER the default tree -- the duplicate factory issue #644 removes."""
+        _write_manifest(tmp_path, [{"id": 10, "path": "main"}])
+        factory = MagicMock(side_effect=RuntimeError("api down"))
+        placement = resolve_scaffold_placement(_project(), tmp_path, 20, factory)
+        assert placement.branch_prefix == "branch-20"
+        assert placement.warning is not None
+        assert "sync pull --branch 20" in placement.warning
+
+    def test_project_mismatch_degrades_with_named_mismatch(self, tmp_path: Path) -> None:
+        """A workspace of a different project: files still land (inert, under
+        branch-{id}/) and the warning names the mismatch -- the remote config
+        already exists, so losing the files entirely would be worse."""
+        _write_manifest(tmp_path, [{"id": 10, "path": "main"}], project_id=9999)
+        placement = resolve_scaffold_placement(_project(1234), tmp_path, 20, MagicMock())
+        assert placement.branch_prefix == "branch-20"
+        assert placement.warning is not None
+        assert "belongs to project 9999" in placement.warning
+
+
+class TestProductionMismatchGuard:
+    def test_production_create_into_foreign_workspace_goes_flat_with_warning(
+        self, tmp_path: Path
+    ) -> None:
+        """A production create pointed at ANOTHER project's workspace must not
+        write into that workspace's main/ tree -- its next sync push would
+        create the config in the WRONG project (PR #653 review sweep). Flat
+        files sit outside every branch tree, so they are inert there."""
+        _write_manifest(tmp_path, [{"id": 10, "path": "main"}], project_id=9999)
+        placement = resolve_scaffold_placement(_project(1234), tmp_path, None, MagicMock())
+        assert placement.branch_prefix is None
+        assert placement.warning is not None
+        assert "belongs to project 9999" in placement.warning
+
+    def test_production_create_matching_project_uses_default_prefix(self, tmp_path: Path) -> None:
+        _write_manifest(tmp_path, [{"id": 10, "path": "main"}], project_id=1234)
+        placement = resolve_scaffold_placement(_project(1234), tmp_path, None, MagicMock())
+        assert placement == ScaffoldPlacement("main")
diff --git a/tests/test_component_service.py b/tests/test_component_service.py
index cdc6ce0d..59e98aab 100644
--- a/tests/test_component_service.py
+++ b/tests/test_component_service.py
@@ -764,3 +764,218 @@ def test_number_type(self) -> None:
         schema = {"properties": {"threshold": {"type": "number", "default": 0.5}}}
         result = _generate_from_schema(schema)
         assert result["threshold"] == 0.5
+
+
+# ===========================================================================
+# stamp_scaffold_config_id + build_pushed_config_files (issue #644)
+# ===========================================================================
+
+
+_SCAFFOLD_YML = (
+    "# Component: HTTP (keboola.ex-http)\n"
+    "# Type: extractor\n"
+    "#\n"
+    "# NOTE: config_id will be assigned by Keboola on first push\n"
+    "version: 2\n"
+    'name: "test-config"\n'
+    "description: |\n"
+    "  TODO: describe this configuration\n"
+    "\n"
+    "parameters: {}\n"
+    "\n"
+    "_keboola:\n"
+    "  component_id: keboola.ex-http\n"
+    "\n"
+)
+
+
+class TestStampScaffoldConfigId:
+    """The --push path must record the created config's ID in the scaffold.
+
+    Without the ID the next ``sync push`` treats the directory as a brand-new
+    configuration and creates a duplicate (issue #644).
+    """
+
+    def _scaffold(self) -> dict[str, Any]:
+        return {
+            "component_id": "keboola.ex-http",
+            "component_name": "HTTP",
+            "component_type": "extractor",
+            "directory": "extractor/keboola.ex-http/test-config",
+            "files": [
+                {"path": "_config.yml", "content": _SCAFFOLD_YML},
+                {"path": "transform.sql", "content": "SELECT 1;\n"},
+            ],
+        }
+
+    def test_stamps_id_into_existing_keboola_block(self) -> None:
+        from keboola_agent_cli.services.component_service import stamp_scaffold_config_id
+
+        result = stamp_scaffold_config_id(self._scaffold(), "01m0njbrqwpyqbx0yqfqq9pyen")
+        content = result["files"][0]["content"]
+        parsed = yaml.safe_load(content)
+        assert parsed["_keboola"]["component_id"] == "keboola.ex-http"
+        assert parsed["_keboola"]["config_id"] == "01m0njbrqwpyqbx0yqfqq9pyen"
+        # The misleading "first push" note must be gone on this path.
+        assert "assigned by Keboola on first push" not in content
+
+    def test_numeric_id_stays_a_string(self) -> None:
+        """Legacy numeric config IDs must round-trip as YAML strings.
+
+        An unquoted ``config_id: 12345`` parses as int and then never matches
+        the string-keyed remote lookup in the sync diff adopt-by-id guard.
+        """
+        from keboola_agent_cli.services.component_service import stamp_scaffold_config_id
+
+        result = stamp_scaffold_config_id(self._scaffold(), "12345")
+        parsed = yaml.safe_load(result["files"][0]["content"])
+        assert parsed["_keboola"]["config_id"] == "12345"
+        assert isinstance(parsed["_keboola"]["config_id"], str)
+
+    def test_appends_block_when_keboola_missing(self) -> None:
+        """Flow scaffolds have no _keboola block (issue #650); stamping must
+        create one so the pushed flow is adoptable too."""
+        from keboola_agent_cli.services.component_service import stamp_scaffold_config_id
+
+        scaffold = self._scaffold()
+        scaffold["files"][0]["content"] = 'name: "my flow"\nphases: []\n'
+        result = stamp_scaffold_config_id(scaffold, "999")
+        parsed = yaml.safe_load(result["files"][0]["content"])
+        assert parsed["_keboola"]["component_id"] == "keboola.ex-http"
+        assert parsed["_keboola"]["config_id"] == "999"
+
+    def test_stamps_real_generator_output_with_single_keboola_block(
+        self, tmp_config_dir: Path
+    ) -> None:
+        """Guard against builder-format drift: feed REAL _build_config_yml
+        output through the stamper (PR #653 review) -- a renamed or
+        reordered _keboola block would silently fall into the append path
+        and produce a duplicate block."""
+        from keboola_agent_cli.services.component_service import stamp_scaffold_config_id
+
+        mock_ai = _make_ai_client(detail_response=EXTRACTOR_RESPONSE)
+        service = _make_service(tmp_config_dir, ai_client=mock_ai)
+        scaffold = service.generate_scaffold(alias="prod", component_id="keboola.ex-http")
+
+        stamped = stamp_scaffold_config_id(scaffold, "01m0njbrqwpyqbx0yqfqq9pyen")
+
+        content = stamped["files"][0]["content"]
+        assert content.count("_keboola:") == 1, "stamping must not append a second block"
+        parsed = yaml.safe_load(content)
+        assert parsed["_keboola"]["component_id"] == "keboola.ex-http"
+        assert parsed["_keboola"]["config_id"] == "01m0njbrqwpyqbx0yqfqq9pyen"
+        assert "assigned by Keboola on first push" not in content
+
+    def test_companion_files_untouched_and_input_not_mutated(self) -> None:
+        from keboola_agent_cli.services.component_service import stamp_scaffold_config_id
+
+        scaffold = self._scaffold()
+        original_yml = scaffold["files"][0]["content"]
+        result = stamp_scaffold_config_id(scaffold, "12345")
+        assert result["files"][1]["content"] == "SELECT 1;\n"
+        # Pure function: the input scaffold must not be mutated.
+        assert scaffold["files"][0]["content"] == original_yml
+
+
+class TestMaterializePushedConfig:
+    """When --configuration was pushed, the local dir must mirror the pushed
+    body exactly the way ``sync pull`` would materialize it -- placeholder
+    scaffolding would make the next ``sync push`` overwrite the real remote
+    config with TODO templates, and inline-only YAML would drop the code
+    files pull-based trees carry."""
+
+    def test_mirrors_pushed_body(self, tmp_path: Path) -> None:
+        from keboola_agent_cli.services.component_service import materialize_pushed_config
+
+        written = materialize_pushed_config(
+            component_id="keboola.ex-http",
+            config_id="12345",
+            name="test-config",
+            description="desc",
+            configuration={
+                "parameters": {
+                    "baseUrl": "https://example.com",
+                    "#token": "KBC::ProjectSecure::abc",
+                },
+                "storage": {"input": {"tables": [{"source": "in.c-b.t"}]}},
+            },
+            config_dir=tmp_path,
+        )
+        # _description.md is pull-parity: sync pull extracts a non-empty
+        # description into a companion file the same way.
+        assert written == ["_config.yml", "_description.md"]
+        parsed = yaml.safe_load((tmp_path / "_config.yml").read_text(encoding="utf-8"))
+        assert parsed["name"] == "test-config"
+        assert parsed["parameters"]["baseUrl"] == "https://example.com"
+        # Encrypted value travels verbatim -- never decrypted, never a TODO.
+        assert parsed["parameters"]["#token"] == "KBC::ProjectSecure::abc"
+        assert parsed["input"] == {"tables": [{"source": "in.c-b.t"}]}
+        assert parsed["_keboola"] == {
+            "component_id": "keboola.ex-http",
+            "config_id": "12345",
+        }
+
+    def test_transformation_body_extracts_real_code(self, tmp_path: Path) -> None:
+        """A pushed SQL body must yield a real transform.sql (like sync pull),
+        NOT a placeholder -- and not disappear entirely (PR #653 review)."""
+        from keboola_agent_cli.services.component_service import materialize_pushed_config
+
+        written = materialize_pushed_config(
+            component_id="keboola.snowflake-transformation",
+            config_id="67890",
+            name="tf",
+            description="",
+            configuration={
+                "parameters": {
+                    "blocks": [
+                        {
+                            "name": "Blocks",
+                            "codes": [{"name": "Code", "script": ["SELECT 1;"]}],
+                        }
+                    ]
+                }
+            },
+            config_dir=tmp_path,
+        )
+        assert "transform.sql" in written and "_config.yml" in written
+        sql = (tmp_path / "transform.sql").read_text(encoding="utf-8")
+        assert "SELECT 1;" in sql
+        assert "TODO" not in sql
+
+    def test_null_parameters_body_does_not_crash(self, tmp_path: Path) -> None:
+        """'{"parameters": null}' is accepted by the Storage API; the local
+        materialization must not crash after the remote create succeeded
+        (PR #653 review sweep)."""
+        from keboola_agent_cli.services.component_service import materialize_pushed_config
+
+        written = materialize_pushed_config(
+            component_id="keboola.snowflake-transformation",
+            config_id="123",
+            name="tf",
+            description="",
+            configuration={"parameters": None},
+            config_dir=tmp_path,
+        )
+        assert "_config.yml" in written
+
+    def test_stale_files_not_reported_and_stale_description_removed(self, tmp_path: Path) -> None:
+        """The slugified dir can pre-exist: stray files must not be reported
+        as written, and a stale _description.md must not misattribute to the
+        new config when the pushed description is empty."""
+        from keboola_agent_cli.services.component_service import materialize_pushed_config
+
+        (tmp_path / "_description.md").write_text("old description", encoding="utf-8")
+        (tmp_path / "leftover_notes.txt").write_text("stray", encoding="utf-8")
+
+        written = materialize_pushed_config(
+            component_id="keboola.ex-http",
+            config_id="123",
+            name="cfg",
+            description="",
+            configuration={"parameters": {"a": 1}},
+            config_dir=tmp_path,
+        )
+        assert written == ["_config.yml"]
+        assert not (tmp_path / "_description.md").exists()
+        # Stray unrelated files are left alone -- just not claimed as ours.
+        assert (tmp_path / "leftover_notes.txt").exists()
diff --git a/tests/test_config_create_cli.py b/tests/test_config_create_cli.py
index 792346b9..6ccea167 100644
--- a/tests/test_config_create_cli.py
+++ b/tests/test_config_create_cli.py
@@ -14,6 +14,7 @@
 from pathlib import Path
 from unittest.mock import MagicMock, patch
 
+import yaml
 from typer.testing import CliRunner, Result
 
 from keboola_agent_cli.cli import app
@@ -21,6 +22,7 @@
 from keboola_agent_cli.errors import ConfigError, KeboolaApiError
 from keboola_agent_cli.models import ProjectConfig
 from keboola_agent_cli.services.project_service import ProjectService
+from keboola_agent_cli.services.sync_service import SyncService
 
 TEST_TOKEN = "901-55555-fakeTestTokenDoNotUseXXXXXXXX"
 
@@ -66,6 +68,7 @@ def _invoke_push(
     *,
     config_service_mock: MagicMock | None = None,
     component_service_mock: MagicMock | None = None,
+    sync_service_mock: MagicMock | None = None,
     config_dir: Path | None = None,
     input_text: str | None = None,
 ) -> Result:
@@ -94,11 +97,13 @@ def _invoke_push(
         patch("keboola_agent_cli.cli.ProjectService") as MockProjService,
         patch("keboola_agent_cli.cli.ComponentService") as MockCompService,
         patch("keboola_agent_cli.cli.ConfigService") as MockConfigService,
+        patch("keboola_agent_cli.cli.SyncService") as MockSyncService,
     ):
         MockStore.return_value = store
         MockProjService.return_value = ProjectService(config_store=store)
         MockCompService.return_value = svc_component
         MockConfigService.return_value = svc_config
+        MockSyncService.return_value = sync_service_mock or SyncService(config_store=store)
         return runner.invoke(app, args, input=input_text)
 
 
@@ -692,3 +697,293 @@ def test_api_error_surfaces_with_mapped_exit_code(self, tmp_path: Path) -> None:
         assert result.exit_code == 1, result.output
         envelope = json.loads(result.output)
         assert envelope["status"] == "error"
+
+
+# ---------------------------------------------------------------------------
+# Scaffold stamping + branch-aware placement (issue #644)
+# ---------------------------------------------------------------------------
+
+_REALISTIC_SCAFFOLD = {
+    "component_id": "keboola.ex-http",
+    "component_name": "HTTP",
+    "component_type": "extractor",
+    "directory": "extractor/keboola.ex-http/test-config",
+    "files": [
+        {
+            "path": "_config.yml",
+            "content": (
+                "# NOTE: config_id will be assigned by Keboola on first push\n"
+                "version: 2\n"
+                'name: "test-config"\n'
+                "parameters: {}\n"
+                "\n"
+                "_keboola:\n"
+                "  component_id: keboola.ex-http\n"
+            ),
+        },
+    ],
+}
+
+
+def _write_manifest(output_dir: Path, branches: list[dict]) -> None:
+    """Materialize a minimal .keboola/manifest.json in *output_dir*."""
+    keboola_dir = output_dir / ".keboola"
+    keboola_dir.mkdir(parents=True, exist_ok=True)
+    manifest = {
+        "version": 2,
+        "project": {"id": 1234, "apiHost": "connection.keboola.com"},
+        "naming": {},
+        "branches": branches,
+        "configurations": [],
+    }
+    (keboola_dir / "manifest.json").write_text(json.dumps(manifest), encoding="utf-8")
+
+
+class TestConfigNewPushScaffoldStamping:
+    """--push --output-dir must write a scaffold that carries the created
+    config's identity; otherwise the next ``sync push`` duplicates it
+    (issue #644)."""
+
+    def _component_mock(self) -> MagicMock:
+        svc = MagicMock()
+        svc.generate_scaffold.return_value = json.loads(json.dumps(_REALISTIC_SCAFFOLD))
+        return svc
+
+    def _run(self, tmp_path: Path, out_dir: Path, extra: list[str] | None = None, **kwargs):
+        args = [
+            "--json",
+            "config",
+            "new",
+            "--component-id",
+            "keboola.ex-http",
+            "--project",
+            "prod",
+            "--name",
+            "test-config",
+            "--push",
+            "--output-dir",
+            str(out_dir),
+        ] + (extra or [])
+        return _invoke_push(args, config_dir=tmp_path / "config", **kwargs)
+
+    def test_written_scaffold_carries_created_config_id(self, tmp_path: Path) -> None:
+        out = tmp_path / "ws"
+        out.mkdir()
+        svc_config = MagicMock()
+        svc_config.create_config.return_value = _push_result()
+
+        result = self._run(
+            tmp_path,
+            out,
+            config_service_mock=svc_config,
+            component_service_mock=self._component_mock(),
+        )
+
+        assert result.exit_code == 0, result.output
+        written = out / "extractor/keboola.ex-http/test-config/_config.yml"
+        parsed = yaml.safe_load(written.read_text(encoding="utf-8"))
+        assert parsed["_keboola"]["config_id"] == "12345"
+        assert isinstance(parsed["_keboola"]["config_id"], str)
+        assert "assigned by Keboola on first push" not in written.read_text(encoding="utf-8")
+
+    def test_envelope_reports_local_scaffold(self, tmp_path: Path) -> None:
+        out = tmp_path / "ws"
+        out.mkdir()
+        svc_config = MagicMock()
+        svc_config.create_config.return_value = _push_result()
+
+        result = self._run(
+            tmp_path,
+            out,
+            config_service_mock=svc_config,
+            component_service_mock=self._component_mock(),
+        )
+
+        envelope = json.loads(result.output)
+        scaffold_info = envelope["data"]["local_scaffold"]
+        assert scaffold_info["files"] == ["_config.yml"]
+        # Path comparison, not str.endswith -- Windows renders backslashes.
+        assert Path(scaffold_info["directory"]) == out / "extractor/keboola.ex-http/test-config"
+
+    def test_branch_create_writes_into_registered_branch_dir(self, tmp_path: Path) -> None:
+        """A config created in a dev branch must scaffold into that branch's
+        subtree, never into the default branch's (main/) tree."""
+        out = tmp_path / "ws"
+        _write_manifest(out, [{"id": 10, "path": "main"}, {"id": 20, "path": "dev-x"}])
+        svc_config = MagicMock()
+        svc_config.create_config.return_value = {**_push_result(), "branch_id": 20}
+
+        result = self._run(
+            tmp_path,
+            out,
+            extra=["--branch", "20"],
+            config_service_mock=svc_config,
+            component_service_mock=self._component_mock(),
+        )
+
+        assert result.exit_code == 0, result.output
+        expected = out / "dev-x/extractor/keboola.ex-http/test-config/_config.yml"
+        assert expected.exists(), (
+            f"expected scaffold under dev-x/, tree: {list(out.rglob('_config.yml'))}"
+        )
+        assert not (out / "main/extractor").exists(), "must not write into the default branch tree"
+
+    def test_branch_unknown_to_manifest_gets_registered(self, tmp_path: Path) -> None:
+        from keboola_agent_cli.sync.branch_registry import ScaffoldPlacement
+
+        out = tmp_path / "ws"
+        _write_manifest(out, [{"id": 10, "path": "main"}])
+        svc_config = MagicMock()
+        svc_config.create_config.return_value = {**_push_result(), "branch_id": 20}
+        sync_mock = MagicMock()
+        sync_mock.resolve_scaffold_placement.return_value = ScaffoldPlacement("issue-branch")
+
+        result = self._run(
+            tmp_path,
+            out,
+            extra=["--branch", "20"],
+            config_service_mock=svc_config,
+            component_service_mock=self._component_mock(),
+            sync_service_mock=sync_mock,
+        )
+
+        assert result.exit_code == 0, result.output
+        sync_mock.resolve_scaffold_placement.assert_called_once()
+        kwargs = sync_mock.resolve_scaffold_placement.call_args.kwargs
+        assert kwargs.get("branch_id") == 20
+        expected = out / "issue-branch/extractor/keboola.ex-http/test-config/_config.yml"
+        assert expected.exists(), f"tree: {list(out.rglob('_config.yml'))}"
+
+    def test_placement_warning_surfaces_in_envelope(self, tmp_path: Path) -> None:
+        """A degraded placement (registration failed -> branch-{id}/) must be
+        honored AND surfaced -- never silently retargeted to the default
+        tree, which is the exact duplicate factory this fix removes."""
+        from keboola_agent_cli.sync.branch_registry import ScaffoldPlacement
+
+        out = tmp_path / "ws"
+        _write_manifest(out, [{"id": 10, "path": "main"}])
+        svc_config = MagicMock()
+        svc_config.create_config.return_value = {**_push_result(), "branch_id": 20}
+        sync_mock = MagicMock()
+        sync_mock.resolve_scaffold_placement.return_value = ScaffoldPlacement(
+            "branch-20", "registration failed; reconcile with sync pull"
+        )
+
+        result = self._run(
+            tmp_path,
+            out,
+            extra=["--branch", "20"],
+            config_service_mock=svc_config,
+            component_service_mock=self._component_mock(),
+            sync_service_mock=sync_mock,
+        )
+
+        assert result.exit_code == 0, result.output
+        expected = out / "branch-20/extractor/keboola.ex-http/test-config/_config.yml"
+        assert expected.exists(), f"tree: {list(out.rglob('_config.yml'))}"
+        assert not (out / "main/extractor").exists()
+        envelope = json.loads(result.output)
+        assert envelope["data"]["warnings"] == ["registration failed; reconcile with sync pull"]
+
+    def test_branch_without_manifest_stays_flat(self, tmp_path: Path) -> None:
+        """No sync workspace in output_dir -> flat layout (placement None)."""
+        from keboola_agent_cli.sync.branch_registry import ScaffoldPlacement
+
+        out = tmp_path / "plain"
+        out.mkdir()
+        svc_config = MagicMock()
+        svc_config.create_config.return_value = {**_push_result(), "branch_id": 20}
+        sync_mock = MagicMock()
+        sync_mock.resolve_scaffold_placement.return_value = ScaffoldPlacement(None)
+
+        result = self._run(
+            tmp_path,
+            out,
+            extra=["--branch", "20"],
+            config_service_mock=svc_config,
+            component_service_mock=self._component_mock(),
+            sync_service_mock=sync_mock,
+        )
+
+        assert result.exit_code == 0, result.output
+        expected = out / "extractor/keboola.ex-http/test-config/_config.yml"
+        assert expected.exists(), f"tree: {list(out.rglob('_config.yml'))}"
+
+    def test_pushed_body_is_mirrored_not_placeholder(self, tmp_path: Path) -> None:
+        """--configuration: the local file mirrors the pushed (already
+        encrypted) body; placeholder scaffolding would make the next push
+        overwrite the real remote config with TODO templates."""
+        out = tmp_path / "ws"
+        out.mkdir()
+        svc_config = MagicMock()
+        svc_config.create_config.return_value = {
+            **_push_result(),
+            "configuration": {"parameters": {"baseUrl": "https://real.example.com"}},
+        }
+
+        result = self._run(
+            tmp_path,
+            out,
+            extra=["--configuration", '{"parameters": {"baseUrl": "https://real.example.com"}}'],
+            config_service_mock=svc_config,
+            component_service_mock=self._component_mock(),
+        )
+
+        assert result.exit_code == 0, result.output
+        written = out / "extractor/keboola.ex-http/test-config/_config.yml"
+        parsed = yaml.safe_load(written.read_text(encoding="utf-8"))
+        assert parsed["parameters"] == {"baseUrl": "https://real.example.com"}
+        assert parsed["_keboola"]["config_id"] == "12345"
+        assert "TODO" not in written.read_text(encoding="utf-8")
+
+    def test_pushed_transformation_body_extracts_code_through_cli(self, tmp_path: Path) -> None:
+        """End-to-end through the Typer command: a pushed SQL body yields a
+        real transform.sql next to _config.yml (PR #653 review sweep -- the
+        mirror branch was previously only unit-tested)."""
+        out = tmp_path / "ws"
+        out.mkdir()
+        body = {
+            "parameters": {
+                "blocks": [{"name": "Blocks", "codes": [{"name": "Code", "script": ["SELECT 1;"]}]}]
+            }
+        }
+        svc_config = MagicMock()
+        svc_config.create_config.return_value = {**_push_result(), "configuration": body}
+        svc_component = MagicMock()
+        svc_component.generate_scaffold.return_value = {
+            **json.loads(json.dumps(_REALISTIC_SCAFFOLD)),
+            "component_id": "keboola.snowflake-transformation",
+            "directory": "transformation/keboola.snowflake-transformation/test-config",
+        }
+
+        result = _invoke_push(
+            [
+                "--json",
+                "config",
+                "new",
+                "--component-id",
+                "keboola.snowflake-transformation",
+                "--project",
+                "prod",
+                "--name",
+                "test-config",
+                "--push",
+                "--output-dir",
+                str(out),
+                "--configuration",
+                json.dumps(body),
+            ],
+            config_dir=tmp_path / "config",
+            config_service_mock=svc_config,
+            component_service_mock=svc_component,
+        )
+
+        assert result.exit_code == 0, result.output
+        base = out / "transformation/keboola.snowflake-transformation/test-config"
+        assert (base / "transform.sql").is_file(), list(out.rglob("*"))
+        assert "SELECT 1;" in (base / "transform.sql").read_text(encoding="utf-8")
+        envelope = json.loads(result.output)
+        assert sorted(envelope["data"]["local_scaffold"]["files"]) == [
+            "_config.yml",
+            "transform.sql",
+        ]
diff --git a/tests/test_e2e.py b/tests/test_e2e.py
index cb14b5db..d2bf96da 100644
--- a/tests/test_e2e.py
+++ b/tests/test_e2e.py
@@ -520,6 +520,9 @@ def test_full_cli_e2e(self) -> None:
         _step("19c", "config new --push validation", "real schema vs real body (#587)")
         self._test_config_new_push_schema_validation()
 
+        _step("19c2", "config new --push --output-dir", "scaffold carries created config_id (#644)")
+        self._test_config_new_push_output_dir()
+
         _step("19d", "config clone", "whole-config duplicate incl. rows (#587)")
         self._test_config_clone()
 
@@ -2198,6 +2201,62 @@ def _test_config_new_push(self) -> None:
                 new_config_id,
             )
 
+    def _test_config_new_push_output_dir(self) -> None:
+        """Test ``config new --push --output-dir`` writes an adoptable scaffold (issue #644).
+
+        The written ``_config.yml`` must carry ``_keboola.config_id`` of the
+        just-created configuration -- before the #644 fix it did not, and the
+        next ``sync push`` created a duplicate (34-config incident). A plain
+        (non-sync) output dir is used, so the layout is flat and no manifest
+        is involved; the stamped ID is the contract under test.
+        """
+        import yaml as _yaml
+
+        push_name = f"{RUN_ID} push-scaffold-644"
+        out_dir = self.work_dir / "scaffold-644"
+        out_dir.mkdir(parents=True, exist_ok=True)
+
+        created = self._run_ok(
+            "config",
+            "new",
+            "--component-id",
+            "keboola.ex-http",
+            "--project",
+            self.alias,
+            "--name",
+            push_name,
+            "--push",
+            "--output-dir",
+            str(out_dir),
+        )["data"]
+        new_config_id = str(created["id"])
+        self._created_config_ids.append(("keboola.ex-http", new_config_id))
+
+        try:
+            scaffold_info = created["local_scaffold"]
+            assert scaffold_info["config_id"] == new_config_id, scaffold_info
+            config_yml = Path(scaffold_info["directory"]) / "_config.yml"
+            assert config_yml.is_file(), f"scaffold not written: {scaffold_info}"
+            raw = config_yml.read_text(encoding="utf-8")
+            parsed = _yaml.safe_load(raw)
+            assert parsed["_keboola"]["component_id"] == "keboola.ex-http"
+            # The stamped ID is the duplicate-prevention contract (#644):
+            # sync diff's adopt-by-id guard pairs the dir with the remote.
+            assert parsed["_keboola"]["config_id"] == new_config_id, raw
+            assert isinstance(parsed["_keboola"]["config_id"], str)
+            assert "assigned by Keboola on first push" not in raw
+        finally:
+            self._run_ok(
+                "config",
+                "delete",
+                "--project",
+                self.alias,
+                "--component-id",
+                "keboola.ex-http",
+                "--config-id",
+                new_config_id,
+            )
+
     def _test_config_new_push_schema_validation(self) -> None:
         """Test ``config new --push`` schema validation against a REAL schema (issue #587).